LongTotalLength

csharp/src/Apache.Arrow.Flight/Internal/FlightDataStream.cs

@@ -91,7 +91,11 @@ public async Task Write(RecordBatch recordBatch, ByteString applicationMetadata)
             await _clientStreamWriter.WriteAsync(_currentFlightData).ConfigureAwait(false);
         }
 
-        private protected override ValueTask<long> WriteMessageAsync<T>(MessageHeader headerType, Offset<T> headerOffset, int bodyLength, CancellationToken cancellationToken)
+        private protected override ValueTask<long> WriteMessageAsync<T>(
+            MessageHeader headerType,
+            Offset<T> headerOffset,
+            long bodyLength,
+            CancellationToken cancellationToken)
         {
             Offset<Flatbuf.Message> messageOffset = Flatbuf.Message.CreateMessage(
                 Builder, CurrentMetadataVersion, headerType, headerOffset.Value,

csharp/src/Apache.Arrow.Flight/Internal/RecordBatchReaderImplementation.cs

@@ -120,7 +120,7 @@ public override async ValueTask<RecordBatch> ReadNextRecordBatchAsync(Cancellati
                 {
                     case MessageHeader.RecordBatch:
                         var body = _flightDataStream.Current.DataBody.Memory;
-                        return CreateArrowObjectFromMessage(message, CreateByteBuffer(body.Slice(0, (int)message.BodyLength)), null);
+                        return CreateArrowObjectFromMessage(message, CreateByteBuffer(body.Slice(0, checked((int)message.BodyLength))), null);
                     default:
                         throw new NotImplementedException();
                 }

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

@@ -83,7 +83,7 @@ public override RecordBatch ReadNextRecordBatch()
                     CreateByteBuffer(_buffer.Slice(_bufferPosition, messageLength)));
                 _bufferPosition += messageLength;
 
-                int bodyLength = (int)message.BodyLength;
+                int bodyLength = checked((int)message.BodyLength);
                 ByteBuffer bodybb = CreateByteBuffer(_buffer.Slice(_bufferPosition, bodyLength));
                 _bufferPosition += bodyLength;
 

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

@@ -121,7 +121,7 @@ protected RecordBatch CreateArrowObjectFromMessage(
                 case Flatbuf.MessageHeader.RecordBatch:
                     Flatbuf.RecordBatch rb = message.Header<Flatbuf.RecordBatch>().Value;
                     List<IArrowArray> arrays = BuildArrays(message.Version, Schema, bodyByteBuffer, rb);
-                    return new RecordBatch(Schema, memoryOwner, arrays, (int)rb.Length);
+                    return new RecordBatch(Schema, memoryOwner, arrays, checked((int)rb.Length));
                 default:
                     // NOTE: Skip unsupported message type
                     Debug.WriteLine($"Skipping unsupported message type '{message.HeaderType}'");
@@ -238,8 +238,8 @@ private ArrayData LoadPrimitiveField(
             IBufferCreator bufferCreator)
         {
 
-            int fieldLength = (int)fieldNode.Length;
-            int fieldNullCount = (int)fieldNode.NullCount;
+            int fieldLength = checked((int)fieldNode.Length);
+            int fieldNullCount = checked((int)fieldNode.NullCount);
 
             if (fieldLength < 0)
             {
@@ -322,8 +322,8 @@ private ArrayData LoadVariableField(
             ArrowBuffer valueArrowBuffer = BuildArrowBuffer(bodyData, recordBatchEnumerator.CurrentBuffer, bufferCreator);
             recordBatchEnumerator.MoveNextBuffer();
 
-            int fieldLength = (int)fieldNode.Length;
-            int fieldNullCount = (int)fieldNode.NullCount;
+            int fieldLength = checked((int)fieldNode.Length);
+            int fieldNullCount = checked((int)fieldNode.NullCount);
 
             if (fieldLength < 0)
             {
@@ -381,8 +381,8 @@ private ArrowBuffer BuildArrowBuffer(ByteBuffer bodyData, Flatbuf.Buffer buffer,
                 return ArrowBuffer.Empty;
             }
 
-            int offset = (int)buffer.Offset;
-            int length = (int)buffer.Length;
+            int offset = checked((int)buffer.Offset);
+            int length = checked((int)buffer.Length);
 
             var data = bodyData.ToReadOnlyMemory(offset, length);
             return bufferCreator.CreateBuffer(data);

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

@@ -68,9 +68,9 @@ internal class ArrowRecordBatchFlatBufferBuilder :
             public readonly struct Buffer
             {
                 public readonly ArrowBuffer DataBuffer;
-                public readonly int Offset;
+                public readonly long Offset;
 
-                public Buffer(ArrowBuffer buffer, int offset)
+                public Buffer(ArrowBuffer buffer, long offset)
                 {
                     DataBuffer = buffer;
                     Offset = offset;
@@ -81,7 +81,7 @@ public Buffer(ArrowBuffer buffer, int offset)
 
             public IReadOnlyList<Buffer> Buffers => _buffers;
 
-            public int TotalLength { get; private set; }
+            public long TotalLength { get; private set; }
 
             public ArrowRecordBatchFlatBufferBuilder()
             {
@@ -210,7 +210,7 @@ private void CreateBuffers<T>(PrimitiveArray<T> array)
 
             private Buffer CreateBuffer(ArrowBuffer buffer)
             {
-                int offset = TotalLength;
+                long offset = TotalLength;
 
                 int paddedLength = checked((int)BitUtility.RoundUpToMultipleOf8(buffer.Length));
                 TotalLength += paddedLength;
@@ -819,7 +819,7 @@ await WriteMessageAsync(Flatbuf.MessageHeader.Schema, schemaOffset, 0, cancellat
         /// The number of bytes written to the stream.
         /// </returns>
         private protected long WriteMessage<T>(
-            Flatbuf.MessageHeader headerType, Offset<T> headerOffset, int bodyLength)
+            Flatbuf.MessageHeader headerType, Offset<T> headerOffset, long bodyLength)
             where T : struct
         {
             Offset<Flatbuf.Message> messageOffset = Flatbuf.Message.CreateMessage(
@@ -849,7 +849,7 @@ private protected long WriteMessage<T>(
         /// The number of bytes written to the stream.
         /// </returns>
         private protected virtual async ValueTask<long> WriteMessageAsync<T>(
-            Flatbuf.MessageHeader headerType, Offset<T> headerOffset, int bodyLength,
+            Flatbuf.MessageHeader headerType, Offset<T> headerOffset, long bodyLength,
             CancellationToken cancellationToken)
             where T : struct
         {

csharp/test/Apache.Arrow.Tests/ArrowFileWriterTests.cs

@@ -14,8 +14,10 @@
 // limitations under the License.
 
 using Apache.Arrow.Ipc;
+using Apache.Arrow.Types;
 using System;
 using System.IO;
+using System.Linq;
 using System.Threading.Tasks;
 using Xunit;
 
@@ -164,5 +166,56 @@ public async Task WritesEmptyFileAsync()
             Assert.Null(readBatch);
             SchemaComparer.Compare(originalBatch.Schema, reader.Schema);
         }
+
+        [Fact]
+        public async Task WritesLargeFileAsync()
+        {
+            int rowCount = 1_000_000;
+            int batchCount = 25;
+            int columns = 25;
+
+            Int64Array manyZeroes = new Int64Array(new ArrowBuffer(new byte[8 * rowCount]), ArrowBuffer.Empty, rowCount, 0, 0);
+            Schema schema = new Schema(Enumerable.Repeat(new Field("", Int64Type.Default, false), columns).ToArray(), null);
+            RecordBatch batch = new RecordBatch(schema, Enumerable.Repeat(manyZeroes, columns), rowCount);
+
+            string tempFile = Path.GetTempFileName();
+            try
+            {
+                using (Stream stream = File.OpenWrite(tempFile))
+                using (ArrowFileWriter writer = new ArrowFileWriter(stream, schema))
+                {
+                    await writer.WriteStartAsync();
+                    for (int i = 0; i < batchCount; i++)
+                    {
+                        await writer.WriteRecordBatchAsync(batch);
+                    }
+                    await writer.WriteEndAsync();
+                }
+
+                using (Stream stream = File.OpenRead(tempFile))
+                using (ArrowFileReader reader = new ArrowFileReader(stream))
+                {
+                    int readCount = await reader.RecordBatchCountAsync();
+                    Assert.Equal(batchCount, readCount);
+
+                    for (int i = 0; i < batchCount; i++)
+                    {
+                        RecordBatch readBatch = await reader.ReadNextRecordBatchAsync();
+                        Assert.Equal(rowCount, readBatch.Length);
+                        Assert.Equal(columns, readBatch.Schema.FieldsList.Count);
+                    }
+                }
+            }
+            finally
+            {
+                try
+                {
+                    File.Delete(tempFile);
+                }
+                catch (Exception)
+                {
+                }
+            }
+        }
     }
 }