Decoupling FirehoseFactory and InputRowParser

[jihoonson]

Original discussion is here.

Firehose is designed to support any file formats, but FirehoseFactory isn't: it should be associated to a specific type of InputRowParser at compile time. This means that we should implement different FirehoseFactorys for each various file formats which has a significant limitation on extensibility.

As intended from the first place, FirehoseFactory should specify how to make a new Firehose specifying how data comes to Druid (like downloading S3 objects from static s3 firehose), while InputRowParser specifies how Druid parses incoming data through Firehose.

[HPCohen]

As I look more into the code for this issue, the big problem seems to be tied to the fact that the firehose needs to determine how to pass an InputRow to the parser, but that can be tied to the file format (a row is stored differently in a csv than in parquet). This entanglement makes it difficult to separate the format from the firehose

[gianm]

I think the real sin is that InputRowParsers receive one "row" at a time, but the responsibility of splitting up a file into "rows" lies with the firehose, and it has to do that without knowing what the input file format is. At that point, what can it do other than something generic like split the file on newlines? Which is what most of the file-oriented FirehoseFactories do. And that means they won't work with Parquet files.

This design made sense for loading data from a message oriented system like Kafka, which is what FirehoseFactories were originally designed to do. It is the responsibility of the message system to deliver one row at a time. But it doesn't really make sense for reading from files, where there is the added complexity of a container format that something needs to be aware of.

I think the right design for this would be that the file-based FirehoseFactories have some additional parameter, "reader" or what have you, that knows how to split streams into rows that are acceptable to the parser. The default should be a LineOrientedReader that splits the stream on newlines into strings, but there could be another one that knows how to read Parquet files. The message-based FirehoseFactories (like Kafka) wouldn't have this parameter.

[jihoonson]

This design made sense for loading data from a message oriented system like Kafka, which is what FirehoseFactories were originally designed to do. It is the responsibility of the message system to deliver one row at a time.

One thing which has changed is, each Kafka record can contain multiple rows now. It is supported since #5081.

[josephglanville]

@jihoonson we actually make use of parse batch support also.

That said InputRowParser should still operate on a single iteration unit IMO.

A Reader interface that returns an iterator seems like the right choice.
It should also remove some weirdness I remember seeing around resetting parsers on starting a new file as you could move that reset to the Reader.

I don't see a reason to not also support Reader at realtime ingestion too, a ReaderInputRowParser that wraps an inner InputRowParser would be an awesome and flexible way to support entire files being pushed down streaming pipelines. Though perhaps integrating it at the level below that makes more sense so that the batch size isn't the entire file.

[jihoonson]

@josephglanville, thanks for the comment. It sounds good to me. I also agree to support Reader for realtime ingestion too. Do you want to make a PR for this?

[josephglanville]

Yes, I am starting work on this now. I will open a PR when I have something we can iterate on.

[josephglanville]

One thing that has become apparent is the StringInputRowParser is a bit of a bad citizen.
It extends ByteBufferInputRowParser aka InputRowParser<ByteBuffer>.

This is not a problem in and of itself and it's completely reasonable to have a input row parser designed to read UTF-8 encoded strings from ByteBuffer.

The issue is that isn't how it's used, it's primarily used via function with signature the InputRow parse(String), the fact this is String rather than ByteBuffer not only violates the assumption of the interface but it also means that users of StringInputRowParser aren't using the InputRowParser interface at all, effectively making any code built on StringInputRowParser not very generic.

It seems to me that it would make sense for there to be StringInputRowParser that is actually InputRowParser<String> and StringDecoderInputRowParser which inherits from ByteBufferInputRowParser reads data from ByteBuffer decoding UTF-8.

I would also move to put add a reset() to the InputRowParser interface as it's already on Parser. Currently there is startFileFromBeginning() on StringInputRowParser that calls the same function on the inner Parser. Seems as this may be useful in general for resetting the state of stateful parsers when opening the next object/input stream.

[gianm]

Yeah -- StringInputRowParser is a bit of a special snowflake. Maybe doing this feature is the motivation we need to rethink how that all works. To add to the points you raised, there's also this gross code in TransformSpec (which decorates InputRowParsers with expression-based transforms):

    if (parser instanceof StringInputRowParser) {
      // Hack to support the fact that some callers use special methods in StringInputRowParser, such as
      // parse(String) and startFileFromBeginning.
      return (InputRowParser<T>) new TransformingStringInputRowParser(
          parser.getParseSpec(),
          ((StringInputRowParser) parser).getEncoding(),
          this
      );
    } else {
      return new TransformingInputRowParser<>(parser, this);
    }

And there's also the warty fact that ParseSpec's makeParser() method is actually only used by StringInputRowParser (and a couple of extension parsers that work by converting inputs to JSON and then parsing them as strings). Other parsers call getTimestampSpec and getDimensionsSpec, but they don't call makeParser. Some parsers, however, can extract useful things from certain types of ParseSpecs. The way this is done is weird, leading to code like this in the Avro extension to yank out an optional FlattenSpec:

    final JSONPathSpec flattenSpec;
    if (parseSpec != null && (parseSpec instanceof AvroParseSpec)) {
      flattenSpec = ((AvroParseSpec) parseSpec).getFlattenSpec();
    } else {
      flattenSpec = JSONPathSpec.DEFAULT;
    }

The only reason StringInputRowParser doesn't have to do something like this is that the method it needs (makeParser) has been granted special status as a member of the ParseSpec interface (even though many ParseSpecs don't implement it, they just throw exceptions).

If we are wanting to be ambitious (I don't see why not!) then I think we can make nicer, clearer interfaces by rethinking how this all works. In a backwards compatible way, since we don't want all of our users to have to rewrite their ingestion specs (which will probably be the tricky part).

@josephglanville what ambition level are you interested in having here? 😄

[jihoonson]

@gianm thanks for the clarifying the requirements. FYI, I'm very ambitious about this!

[gianm]

Has anyone been working on this? I'm starting to think about it again, in the context of wanting regular "index" tasks to be able to support stuff like Parquet files.

[josephglanville]

I have done some work privately on our usecase to support our batch ingestion pipeline but I have been snowed under with other stuff to get it into a state that would be usable for anyone else.
The primary sticking point is still that StringInputRowParser is a very hairy problem that I don't have the bandwidth to tackle.

[josephglanville]

What I came up with was this:

Firstly a new interface to handle reading a file and emitting rows that can be passed to parsers:

public interface FirehoseReader<T>
{
	boolean hasNext();
	T next();
	void close();
}

For our case we implement a FirehoseReader that reads sequence files and emits ByteBuffer so that any ByteBufferInputRowParser implementations can be used with it:

public class SequenceFileFirehoseReader implements FirehoseReader<ByteBuffer> {
	private Reader reader;
	private Closeable closer;

	public SequenceFileFirehoseReader(InputStream inputStream) throws IOException {
		this(inputStream, null);
	}

	public SequenceFileFirehoseReader(InputStream inputStream, Closeable closer) throws IOException {
		Configuration config = new Configuration();
		config.set("io.compression.codecs", "io.airlift.compress.snappy.SnappyCodec");
		FSDataInputStream fsDataInputStream = new FSDataInputStream(inputStream);
		Option isOption = Reader.stream(fsDataInputStream);
		this.reader = new Reader(config, isOption);
		this.closer = closer;
	}

	@Override
	public boolean hasNext() {
		try {
			Writable ignoreKey = (Writable) reader.getKeyClass().newInstance();
			return reader.next(ignoreKey);
		} catch (IOException | InstantiationException | IllegalAccessException e) {
			return false;
		}
	}

	@Override
	public ByteBuffer next() {
		BytesWritable value = new BytesWritable();
		try {
			reader.getCurrentValue(value);
		} catch (IOException ignored) {
			return null;
		}
		return ByteBuffer.wrap(value.getBytes(), 0, value.getLength());
	}

	@Override
	public void close() {
		try {
			reader.close();
		} catch (IOException ignored) {
		}
		try {
			closer.close();
		} catch (IOException ignored) {
		}
	}
}

This is then coupled with ReaderIteratingFirehose which can be returned from FirehoseFactory.connect()

public class ReaderIteratingFirehose<T> implements Firehose {
	private final Iterator<FirehoseReader<T>> readers;
	private final Closeable closer;
	private FirehoseReader<T> reader = null;
	private InputRowParser<T> parser;
	private Iterator<InputRow> rowIterator;

	public ReaderIteratingFirehose(Iterator<FirehoseReader<T>> readers, InputRowParser<T> parser) {
		this(readers, parser, null);
	}

	public ReaderIteratingFirehose(
		Iterator<FirehoseReader<T>> readers,
		InputRowParser<T> parser,
		Closeable closer
	) {
		this.readers = readers;
		this.parser = parser;
		this.closer = closer;
	}

	@Override
	public boolean hasMore() {
		if (rowIterator != null && rowIterator.hasNext()) {
			return true;
		}
		while ((reader == null || !reader.hasNext()) && readers.hasNext()) {
			reader = getNextReader();
		}

		return reader != null && reader.hasNext();
	}

	@Nullable
	@Override
	public InputRow nextRow() {
		if (!hasMore()) {
			throw new NoSuchElementException();
		}

		// If we don't have any more parsed rows available parse a new batch
		if (rowIterator == null || !rowIterator.hasNext()) {
			rowIterator = parser.parseBatch(reader.next()).iterator();
		}

		return rowIterator.next();
	}

	private FirehoseReader<T> getNextReader() {
		if (reader != null) {
			reader.close();
		}

		return readers.next();
	}

	@Override
	public Runnable commit() {
		return Runnables.getNoopRunnable();
	}

	@Override
	public void close() throws IOException {
		try {
			if (reader != null) {
				reader.close();
			}
		} catch (Throwable t) {
			try {
				if (closer != null) {
					closer.close();
				}
			} catch (Exception e) {
				t.addSuppressed(e);
			}
			throw t;
		}
		if (closer != null) {
			closer.close();
		}
	}
}

For example to use it with the SequenceFileFirehoseReader above:

		return ReaderIteratingFirehose<ByteBuffer>(
			object : Iterator<SequenceFileFirehoseReader> {
				override fun hasNext(): Boolean {
					return fetcher.hasNext()
				}

				override fun next(): SequenceFileFirehoseReader {
					if (!hasNext()) {
						throw NoSuchElementException()
					}

					val openedObject = fetcher.next()
					val stream: InputStream
					try {
						stream = wrapObjectStream(
							openedObject,
							openObjectStream(openedObject)
						)
					} catch (e: IOException) {
						throw RuntimeException(e)
					}

					return SequenceFileFirehoseReader(stream)
				}
			},
			parser
		)

There is still a lot of work to be done though. Beyond fixing up StringInputRowParser the prefetching and file caching firehose stuff would need to be refactored etc.

I should have some more bandwith for this soon and any mentoring/advice on how to deal with the StringInputRowParser situation would be greatly appreciated as we would like to get this work upstream.

[gianm]

Thanks @josephglanville - what have been the toughest issues you faced with StringInputRowParser so far? I haven't scoped out the work that would be needed there yet, so maybe starting with what you've already noticed would be helpful.