CurtHagenlocher commented on code in PR #340:
URL: https://github.com/apache/arrow-dotnet/pull/340#discussion_r3154395445
##########
test/Apache.Arrow.Tests/ArrowStreamReaderTests.cs:
##########
@@ -315,163 +609,6 @@ [new Field("index", Int32Type.Default, nullable: false)],
});
}
- [Fact]
Review Comment:
I think this is a merge error. I recently added these tests and there's no
reason to delete them again :D.
##########
src/Apache.Arrow/Ipc/ArrowMemoryStreamReaderImplementation.cs:
##########
@@ -0,0 +1,212 @@
+// 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;
+using System.Buffers;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using Apache.Arrow.Memory;
+
+namespace Apache.Arrow.Ipc
+{
+ /// <summary>
+ /// Reads Arrow IPC streams from a <see cref="MemoryStream"/> whose
backing buffer is publicly visible.
+ /// </summary>
+ /// <remarks>
+ /// Message metadata can be read directly from the exposed stream buffer,
but record batch bodies are
+ /// still copied into allocator-owned buffers to preserve <see
cref="ArrowStreamReader"/> ownership semantics.
+ /// </remarks>
+ internal sealed class ArrowMemoryStreamReaderImplementation :
ArrowStreamReaderImplementation
+ {
+ private readonly MemoryStream _stream;
+
+ public ArrowMemoryStreamReaderImplementation(
+ MemoryStream stream,
+ MemoryAllocator allocator,
+ ICompressionCodecFactory compressionCodecFactory,
+ bool leaveOpen,
+ ExtensionTypeRegistry extensionRegistry)
+ : base(stream, allocator, compressionCodecFactory, leaveOpen,
extensionRegistry)
+ {
+ _stream = stream;
Review Comment:
Consider extracting the `Memory<byte>` in the constructor and reusing it
instead of extracting it each time. Otherwise, every time it's extracted it
allocates a new object.
##########
src/Apache.Arrow/Ipc/ArrowMemoryStreamReaderImplementation.cs:
##########
@@ -0,0 +1,212 @@
+// 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;
+using System.Buffers;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using Apache.Arrow.Memory;
+
+namespace Apache.Arrow.Ipc
+{
+ /// <summary>
+ /// Reads Arrow IPC streams from a <see cref="MemoryStream"/> whose
backing buffer is publicly visible.
+ /// </summary>
+ /// <remarks>
+ /// Message metadata can be read directly from the exposed stream buffer,
but record batch bodies are
+ /// still copied into allocator-owned buffers to preserve <see
cref="ArrowStreamReader"/> ownership semantics.
+ /// </remarks>
+ internal sealed class ArrowMemoryStreamReaderImplementation :
ArrowStreamReaderImplementation
+ {
+ private readonly MemoryStream _stream;
+
+ public ArrowMemoryStreamReaderImplementation(
+ MemoryStream stream,
+ MemoryAllocator allocator,
+ ICompressionCodecFactory compressionCodecFactory,
+ bool leaveOpen,
+ ExtensionTypeRegistry extensionRegistry)
+ : base(stream, allocator, compressionCodecFactory, leaveOpen,
extensionRegistry)
+ {
+ _stream = stream;
+ }
+
+ public override ValueTask<RecordBatch>
ReadNextRecordBatchAsync(CancellationToken cancellationToken)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+
+ try
+ {
+ return new ValueTask<RecordBatch>(ReadNextRecordBatch());
+ }
+ catch (Exception ex)
+ {
+ return new
ValueTask<RecordBatch>(Task.FromException<RecordBatch>(ex));
+ }
+ }
+
+ public override RecordBatch ReadNextRecordBatch()
+ {
+ ReadSchema();
+
+ ReadResult result = default;
+ do
+ {
+ result = ReadMessageFromExposedMemoryStream();
+ } while (result.Batch == null && result.MessageLength > 0);
+
+ return result.Batch;
+ }
+
+ public override ValueTask<Schema> ReadSchemaAsync(CancellationToken
cancellationToken = default)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+
+ if (HasReadSchema)
+ {
+ return new ValueTask<Schema>(_schema);
+ }
+
+ try
+ {
+ ReadSchema();
+ return new ValueTask<Schema>(_schema);
+ }
+ catch (Exception ex)
+ {
+ return new ValueTask<Schema>(Task.FromException<Schema>(ex));
+ }
+ }
+
+ public override void ReadSchema()
+ {
+ if (HasReadSchema)
+ {
+ return;
+ }
+
+ int schemaMessageLength =
ReadMessageLengthFromExposedMemoryStream(throwOnFullRead: true,
returnOnEmptyStream: true);
+ if (schemaMessageLength == 0)
+ {
+ return;
+ }
+
+ Memory<byte> schemaBuffer = ReadExposedMemory(schemaMessageLength);
+ _schema =
MessageSerializer.GetSchema(ReadMessage<Flatbuf.Schema>(CreateByteBuffer(schemaBuffer)),
ref _dictionaryMemo, _extensionRegistry);
+ }
+
+ private ReadResult ReadMessageFromExposedMemoryStream()
Review Comment:
Consider making the method names more terse now that all the methods work
against a `MemoryStream`
##########
src/Apache.Arrow/Ipc/ArrowStreamReaderImplementation.cs:
##########
@@ -47,11 +47,9 @@ protected override void Dispose(bool disposing)
}
}
- public override async ValueTask<RecordBatch>
ReadNextRecordBatchAsync(CancellationToken cancellationToken)
+ public override ValueTask<RecordBatch>
ReadNextRecordBatchAsync(CancellationToken cancellationToken)
{
- // TODO: Loop until a record batch is read.
- cancellationToken.ThrowIfCancellationRequested();
- return await
ReadRecordBatchAsync(cancellationToken).ConfigureAwait(false);
+ return ReadRecordBatchAsync(cancellationToken);
Review Comment:
Perhaps the `ThrowIfCancellationRequested` should be put back? (I don't know
that it's needed, but there doesn't seem to be a good reason to remove it.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
