apacheGH-43672: [C#] Schema should be optional on FlightInfo (apache#…

…43673)

### Rationale for this change

Schema is not required on a FlightInfo message and sometimes needs to be lazily evaluated on the server. This PR allows schema to be null on the FlightInfo since it will be picked up later when requests with those tickets are made.

### What changes are included in this PR?

### Are these changes tested?
Yes, added a test to confirm this behaviour

### Are there any user-facing changes?

* GitHub Issue: apache#43672

Authored-by: neilglover <neilglover@gmail.com>
Signed-off-by: Curt Hagenlocher <curt@hagenlocher.org>

csharp/src/Apache.Arrow.Flight/FlightInfo.cs

@@ -18,14 +18,15 @@
 using System.Text;
 using Apache.Arrow.Flight.Internal;
 using Apache.Arrow.Ipc;
+using Google.Protobuf;
 
 namespace Apache.Arrow.Flight
 {
  public class FlightInfo
  {
  internal FlightInfo(Protocol.FlightInfo flightInfo)
  {
- Schema = FlightMessageSerializer.DecodeSchema(flightInfo.Schema.Memory);
+ Schema = flightInfo.Schema?.Length > 0 ? FlightMessageSerializer.DecodeSchema(flightInfo.Schema.Memory) : null;
  Descriptor = new FlightDescriptor(flightInfo.FlightDescriptor);
 
  var endpoints = new List<FlightEndpoint>();
@@ -60,7 +61,7 @@ public FlightInfo(Schema schema, FlightDescriptor descriptor, IReadOnlyList<Flig
 
  internal Protocol.FlightInfo ToProtocol()
  {
- var serializedSchema = SchemaWriter.SerializeSchema(Schema);
+ var serializedSchema = Schema != null ? SchemaWriter.SerializeSchema(Schema) : ByteString.Empty;
  var response = new Protocol.FlightInfo()
  {
  Schema = serializedSchema,

csharp/test/Apache.Arrow.Flight.Tests/FlightTests.cs

@@ -71,14 +71,14 @@ private FlightInfo GivenStoreBatches(FlightDescriptor flightDescriptor, params R
  {
  var initialBatch = batches.FirstOrDefault();
 
- var flightHolder = new FlightHolder(flightDescriptor, initialBatch.RecordBatch.Schema, _testWebFactory.GetAddress());
+ var flightHolder = new FlightHolder(flightDescriptor, initialBatch?.RecordBatch.Schema, _testWebFactory.GetAddress());
 
  foreach (var batch in batches)
  {
  flightHolder.AddBatch(batch);
  }
 
- _flightStore.Flights.Add(flightDescriptor, flightHolder);
+ _flightStore.Flights[flightDescriptor] = flightHolder;
 
  return flightHolder.GetFlightInfo();
  }
@@ -124,6 +124,31 @@ public async Task TestPutTwoRecordBatches()
  ArrowReaderVerifier.CompareBatches(expectedBatch2, actualBatches[1].RecordBatch);
  }
 
+ [Fact]
+ public async Task TestGetRecordBatchWithDelayedSchema()
+ {
+ var flightDescriptor = FlightDescriptor.CreatePathDescriptor("test");
+ var expectedBatch = CreateTestBatch(0, 100);
+
+ //Add flight info only to the in memory store without schema or batch
+ GivenStoreBatches(flightDescriptor);
+
+ //Get the flight info for the ticket and verify the schema is null
+ var flightInfo = await _flightClient.GetInfo(flightDescriptor);
+ Assert.Single(flightInfo.Endpoints);
+ Assert.Null(flightInfo.Schema);
+
+ var endpoint = flightInfo.Endpoints.FirstOrDefault();
+
+ //Update the store with the batch and schema
+ GivenStoreBatches(flightDescriptor, new RecordBatchWithMetadata(expectedBatch));
+ var getStream = _flightClient.GetStream(endpoint.Ticket);
+ var resultList = await getStream.ResponseStream.ToListAsync();
+
+ Assert.Single(resultList);
+ ArrowReaderVerifier.CompareBatches(expectedBatch, resultList[0]);
+ }
+
  [Fact]
  public async Task TestGetSingleRecordBatch()
  {