Expose DataStreamWriter.Foreach API

[suhsteve]

This PR exposes the DataStreamWriter.Foreach API.

#208

Users can use this API by creating a [Serializable] class that implements the following interface:

    public interface IForeachWriter
    {
        bool Open(long partitionId, long epochId);

        void Process(Row row);

        void Close(Exception errorOrNull);
    }

This user-defined class will be wrapped in a Wrapper that will call its respective methods according to the spark ForeachWriter lifecycle specifications. The lifecycle of the methods are as follows:

For each partition with partitionId:
    For each batch/epoch of streaming data(if its streaming query) with epochId:
        Method Open(partitionId, epochId) is called.
        If Open returns true:
            For each row in the partition and batch/epoch, method Process(row) is called.
        Method Close(errorOrNull) is called with error(if any) seen while processing rows.

An example of a IForeachWriter can be something like the following:

        [Serializable]
        private class TestForeachWriter : IForeachWriter
        {
            [NonSerialized]
            private StreamWriter _streamWriter;

            private readonly string _writePath;

            public TestForeachWriter(string writePath)
            {
                _writePath = writePath;
            }

            public void Close(Exception errorOrNull)
            {
                _streamWriter?.Dispose();
            }

            public bool Open(long partitionId, long epochId)
            {
                try
                {
                    _streamWriter= new StreamWriter(
                        Path.Combine(
                            _writePath,
                            $"sink-foreachWriter-{Guid.NewGuid()}.csv"));
                    return true;
                }
                catch
                {
                    return false;
                }
            }

            public void Process(Row value)
            {
                _streamWriter.WriteLine(string.Join(",", value.Values));
            }

[imback82]

Can we just remove this side effect? You can use locks such that there is always only one StreamWriter per file.

[imback82]

I also see this comment A single instance of this class is responsible of all the data generated by a single task. Do we need to handle thread issue here?

[suhsteve]

Originally I this a static field because the StreamWriter was not marked as [Serializable]. But since we are creating the StreamWriter object in the open call, marking the field as [NonSerialized] addresses the issue.

[suhsteve]

I also see this comment A single instance of this class is responsible of all the data generated by a single task. Do we need to handle thread issue here?

If all the fields of the ForeachWriter are non static then I don't think there will be issues, but if there are static fields then I think it'll be up to the user to determine how to handle them.

[imback82]

I have few questions / comments, but generally looks good to me.

[imback82]

very minor comment / question.

[imback82]

LGTM. Thanks @suhsteve!

[imback82]

@elvaliuliuliu we need to update the breaking change section in release notes with this change.