Composite fixes (#5553)

(cherry picked from commit fb2bceb)

# Conflicts:
#	src/Npgsql/Internal/PgWriter.cs

src/Npgsql/Internal/Composites/Metadata/CompositeFieldInfo.cs

@@ -37,6 +37,7 @@ private protected CompositeFieldInfo(string name, PgTypeInfo typeInfo, PgTypeId
                 return;
             }
             _binaryBufferRequirements = bufferRequirements;
+            Converter = resolution.Converter;
         }
     }
 
@@ -48,7 +49,6 @@ public PgConverter GetReadInfo(out Size readRequirement)
             return Converter;
         }
 
-        // TODO this is effectively static work, we could optimize this away.
         if (!PgTypeInfo.TryBind(new Field(Name, PgTypeInfo.PgTypeId.GetValueOrDefault(), -1), DataFormat.Binary, out var converterInfo))
             ThrowHelper.ThrowInvalidOperationException("Converter must support binary format to participate in composite types.");
 

src/Npgsql/Internal/Converters/CompositeConverter.cs

@@ -28,27 +28,19 @@ public CompositeConverter(CompositeInfo<T> composite)
                 writeReq = writeReq.Combine(Size.CreateUpperBound(0));
             }
 
-            req = req.Combine(
-                // If a read is Unknown (streaming) we can map it to zero as we just want a minimum buffered size.
-                readReq is { Kind: SizeKind.Unknown } ? Size.Zero : readReq,
-                // For writes Unknown means our size is dependent on the value so we can't ignore it.
-                writeReq);
+            req = req.Combine(readReq, writeReq);
         }
 
         // We have to put a limit on the requirements we report otherwise smaller buffer sizes won't work.
         req = BufferRequirements.Create(Limit(req.Read), Limit(req.Write));
 
         _bufferRequirements = req;
 
+        // Return unknown if we hit the limit.
         Size Limit(Size requirement)
         {
             const int maxByteCount = 1024;
-            return requirement switch
-            {
-                { Kind: SizeKind.UpperBound } => Size.CreateUpperBound(Math.Min(maxByteCount, requirement.Value)),
-                { Kind: SizeKind.Exact } => Size.Create(Math.Min(maxByteCount, requirement.Value)),
-                _ => Size.Unknown
-            };
+            return requirement.GetValueOrDefault() > maxByteCount ? requirement.Combine(Size.Unknown) : requirement;
         }
     }
 

src/Npgsql/Internal/PgWriter.cs

@@ -461,19 +461,20 @@ internal ValueTask Flush(bool async, CancellationToken cancellationToken = defau
     internal ValueTask<NestedWriteScope> BeginNestedWrite(bool async, Size bufferRequirement, int byteCount, object? state, CancellationToken cancellationToken)
     {
         Debug.Assert(bufferRequirement != -1);
-        if (ShouldFlush(bufferRequirement))
-            return Core(async, bufferRequirement, byteCount, state, cancellationToken);
 
+        // ShouldFlush depends on the current size for upper bound requirements, so we must set it beforehand.
         _current = new() { Format = _current.Format, Size = byteCount, BufferRequirement = bufferRequirement, WriteState = state };
 
+        if (ShouldFlush(bufferRequirement))
+            return Core(async, cancellationToken);
+
         return new(new NestedWriteScope());
 #if NET6_0_OR_GREATER
         [AsyncMethodBuilder(typeof(PoolingAsyncValueTaskMethodBuilder<>))]
 #endif
-        async ValueTask<NestedWriteScope> Core(bool async, Size bufferRequirement, int byteCount, object? state, CancellationToken cancellationToken)
+        async ValueTask<NestedWriteScope> Core(bool async, CancellationToken cancellationToken)
         {
             await Flush(async, cancellationToken).ConfigureAwait(false);
-            _current = new() { Format = _current.Format, Size = byteCount, BufferRequirement = bufferRequirement, WriteState = state };
             return new();
         }
     }

test/Npgsql.Tests/Types/CompositeHandlerTests.Read.cs

@@ -19,11 +19,15 @@ async Task Read<T>(T composite, Action<Func<T>, T> assert, string? schema = null
     {
         await using var dataSource = await OpenAndMapComposite(composite, schema, nameof(Read), out var name);
         await using var connection = await dataSource.OpenConnectionAsync();
-        await using var command = new NpgsqlCommand($"SELECT ROW({composite.GetValues()})::{name}", connection);
+
+        var literal = $"ROW({composite.GetValues()})::{name}";
+        var arrayLiteral = $"ARRAY[{literal}]::{name}[]";
+        await using var command = new NpgsqlCommand($"SELECT {literal}, {arrayLiteral}", connection);
         await using var reader = command.ExecuteReader();
 
         await reader.ReadAsync();
         assert(() => reader.GetFieldValue<T>(0), composite);
+        assert(() => reader.GetFieldValue<T[]>(1)[0], composite);
     }
 
     [Test]

test/Npgsql.Tests/Types/CompositeHandlerTests.Write.cs

@@ -19,14 +19,28 @@ async Task Write<T>(T composite, Action<NpgsqlDataReader, T>? assert = null, str
     {
         await using var dataSource = await OpenAndMapComposite(composite, schema, nameof(Write), out var _);
         await using var connection = await dataSource.OpenConnectionAsync();
-        await using var command = new NpgsqlCommand("SELECT (@c).*", connection);
+        {
+            await using var command = new NpgsqlCommand("SELECT (@c).*", connection);
+
+            command.Parameters.AddWithValue("c", composite);
+            await using var reader = await command.ExecuteReaderAsync();
+            await reader.ReadAsync();
+
+            if (assert is not null)
+                assert(reader, composite);
+        }
+
+        {
+            await using var command = new NpgsqlCommand("SELECT (@arrayc)[1].*", connection);
+
+            command.Parameters.AddWithValue("arrayc", new[] { composite });
+            await using var reader = await command.ExecuteReaderAsync();
+            await reader.ReadAsync();
 
-        command.Parameters.AddWithValue("c", composite);
-        await using var reader = await command.ExecuteReaderAsync();
-        await reader.ReadAsync();
 
-        if (assert is not null)
-            assert(reader, composite);
+            if (assert is not null)
+                assert(reader, composite);
+        }
     }
 
     [Test]