ARROW-5034: [C#] ArrowStreamWriter and ArrowFileWriter implement sync WriteRecordBatch

csharp/src/Apache.Arrow/Apache.Arrow.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <TargetFrameworks>netstandard1.3;netcoreapp2.1</TargetFrameworks>
@@ -37,5 +37,6 @@
   </ItemGroup>
   <ItemGroup Condition="'$(TargetFramework)' == 'netcoreapp2.1'">
     <Compile Remove="Extensions\StreamExtensions.netstandard.cs" />
+    <Compile Remove="Extensions\TupleExtensions.netstandard.cs" />
   </ItemGroup>
 </Project>

csharp/src/Apache.Arrow/Extensions/StreamExtensions.netcoreapp2.1.cs

@@ -25,5 +25,10 @@ public static int Read(this Stream stream, Memory<byte> buffer)
         {
             return stream.Read(buffer.Span);
         }
+
+        public static void Write(this Stream stream, ReadOnlyMemory<byte> buffer)
+        {
+            stream.Write(buffer.Span);
+        }
     }
 }

csharp/src/Apache.Arrow/Extensions/StreamExtensions.netstandard.cs

@@ -74,6 +74,27 @@ async ValueTask<int> FinishReadAsync(Task<int> readTask, byte[] localBuffer, Mem
             }
         }
 
+        public static void Write(this Stream stream, ReadOnlyMemory<byte> buffer)
+        {
+            if (MemoryMarshal.TryGetArray(buffer, out ArraySegment<byte> array))
+            {
+                stream.Write(array.Array, array.Offset, array.Count);
+            }
+            else
+            {
+                byte[] sharedBuffer = ArrayPool<byte>.Shared.Rent(buffer.Length);
+                try
+                {
+                    buffer.Span.CopyTo(sharedBuffer);
+                    stream.Write(sharedBuffer, 0, buffer.Length);
+                }
+                finally
+                {
+                    ArrayPool<byte>.Shared.Return(sharedBuffer);
+                }
+            }
+        }
+
         public static ValueTask WriteAsync(this Stream stream, ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
         {
             if (MemoryMarshal.TryGetArray(buffer, out ArraySegment<byte> array))

csharp/src/Apache.Arrow/Extensions/TupleExtensions.netstandard.cs

@@ -0,0 +1,29 @@
+// 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.
+
+using System;
+
+namespace Apache.Arrow
+{
+    // Helpers to Deconstruct Tuples on netstandard
+    internal static partial class TupleExtensions
+    {
+        public static void Deconstruct<T1, T2>(this Tuple<T1, T2> value, out T1 item1, out T2 item2)
+        {
+            item1 = value.Item1;
+            item2 = value.Item2;
+        }
+    }
+}

csharp/src/Apache.Arrow/Ipc/ArrowFileWriter.cs

@@ -61,6 +61,19 @@ public ArrowFileWriter(Stream stream, Schema schema, bool leaveOpen, IpcOptions
             RecordBatchBlocks = new List<Block>();
         }
 
+        public override void WriteRecordBatch(RecordBatch recordBatch)
+        {
+            // TODO: Compare record batch schema
+
+            if (!HasWrittenHeader)
+            {
+                WriteHeader();
+                HasWrittenHeader = true;
+            }
+
+            WriteRecordBatchInternal(recordBatch);
+        }
+
         public override async Task WriteRecordBatchAsync(RecordBatch recordBatch, CancellationToken cancellationToken = default)
         {
             // TODO: Compare record batch schema
@@ -104,13 +117,28 @@ private protected override void FinishedWritingRecordBatch(long bodyLength, long
             _currentRecordBatchOffset = -1;
         }
 
+        private protected override void WriteEndInternal()
+        {
+            base.WriteEndInternal();
+
+            WriteFooter(Schema);
+        }
+
         private protected override async ValueTask WriteEndInternalAsync(CancellationToken cancellationToken)
         {
             await base.WriteEndInternalAsync(cancellationToken);
 
             await WriteFooterAsync(Schema, cancellationToken);
         }
 
+        private void WriteHeader()
+        {
+            // Write magic number and empty padding up to the 8-byte boundary
+
+            WriteMagic();
+            WritePadding(CalculatePadding(ArrowFileConstants.Magic.Length));
+        }
+
         private async Task WriteHeaderAsync(CancellationToken cancellationToken)
         {
             // Write magic number and empty padding up to the 8-byte boundary
@@ -120,6 +148,64 @@ await WritePaddingAsync(CalculatePadding(ArrowFileConstants.Magic.Length))
                 .ConfigureAwait(false);
         }
 
+        private void WriteFooter(Schema schema)
+        {
+            Builder.Clear();
+
+            long offset = BaseStream.Position;
+
+            // Serialize the schema
+
+            FlatBuffers.Offset<Flatbuf.Schema> schemaOffset = SerializeSchema(schema);
+
+            // Serialize all record batches
+
+            Flatbuf.Footer.StartRecordBatchesVector(Builder, RecordBatchBlocks.Count);
+
+            foreach (Block recordBatch in RecordBatchBlocks)
+            {
+                Flatbuf.Block.CreateBlock(
+                    Builder, recordBatch.Offset, recordBatch.MetadataLength, recordBatch.BodyLength);
+            }
+
+            FlatBuffers.VectorOffset recordBatchesVectorOffset = Builder.EndVector();
+
+            // Serialize all dictionaries
+            // NOTE: Currently unsupported.
+
+            Flatbuf.Footer.StartDictionariesVector(Builder, 0);
+
+            FlatBuffers.VectorOffset dictionaryBatchesOffset = Builder.EndVector();
+
+            // Serialize and write the footer flatbuffer
+
+            FlatBuffers.Offset<Flatbuf.Footer> footerOffset = Flatbuf.Footer.CreateFooter(Builder, CurrentMetadataVersion,
+                schemaOffset, dictionaryBatchesOffset, recordBatchesVectorOffset);
+
+            Builder.Finish(footerOffset.Value);
+
+            WriteFlatBuffer();
+
+            // Write footer length
+
+            Buffers.RentReturn(4, (buffer) =>
+            {
+                int footerLength;
+                checked
+                {
+                    footerLength = (int)(BaseStream.Position - offset);
+                }
+
+                BinaryPrimitives.WriteInt32LittleEndian(buffer.Span, footerLength);
+
+                BaseStream.Write(buffer);
+            });
+
+            // Write magic
+
+            WriteMagic();
+        }
+
         private async Task WriteFooterAsync(Schema schema, CancellationToken cancellationToken)
         {
             Builder.Clear();
@@ -182,6 +268,11 @@ await Buffers.RentReturnAsync(4, async (buffer) =>
             await WriteMagicAsync(cancellationToken).ConfigureAwait(false);
         }
 
+        private void WriteMagic()
+        {
+            BaseStream.Write(ArrowFileConstants.Magic);
+        }
+
         private ValueTask WriteMagicAsync(CancellationToken cancellationToken)
         {
             return BaseStream.WriteAsync(ArrowFileConstants.Magic, cancellationToken);