ArgumentNullException when reading empty/null PLP strings with new async behavior

src/Microsoft.Data.SqlClient/tests/ManualTests/SQL/DataReaderTest/DataReaderTest.cs

@@ -1123,5 +1123,114 @@ [Value] [nvarchar](max) NULL
                 }
             }
         }
+
+        // Test for Issue #593 - ArgumentNullException when reading empty/null PLP strings with new async behavior
+        [ConditionalFact(typeof(DataTestUtility), nameof(DataTestUtility.AreConnStringsSetup))]
+        public static async Task ReadEmptyAndNullPlpStringsAsyncWithNewBehavior()
+        {
+            string tableName = DataTestUtility.GenerateObjectName();
+
+            try
+            {
+                // Disable compatibility mode to use new async behavior.
+                // SwitchesHelper restores original values automatically on dispose.
+                using (var switchHelper = new SwitchesHelper())
+                {
+                    switchHelper.UseCompatibilityAsyncBehaviour = false;
+
+                    using (SqlConnection connection = new SqlConnection(DataTestUtility.TCPConnectionString))
+                    {
+                        await connection.OpenAsync();
+
+                        // Create test table with large string columns (VARCHAR(MAX), NVARCHAR(MAX) are PLP types; TEXT is a deprecated LOB type)
+                        using (SqlCommand cmd = connection.CreateCommand())
+                        {
+                            cmd.CommandText = $@"
+                                CREATE TABLE [{tableName}] (
+                                    Id INT PRIMARY KEY,
+                                    VarcharMaxCol VARCHAR(MAX),
+                                    NVarcharMaxCol NVARCHAR(MAX),
+                                    TextCol TEXT
+                                )";
+                            await cmd.ExecuteNonQueryAsync();
+                        }
+
+                        // Insert test data: NULL, empty string, and non-empty values
+                        using (SqlCommand cmd = connection.CreateCommand())
+                        {
+                            cmd.CommandText = $@"
+                                INSERT INTO [{tableName}] (Id, VarcharMaxCol, NVarcharMaxCol, TextCol) VALUES
+                                (1, NULL, NULL, NULL),
+                                (2, '', '', ''),
+                                (3, 'test', N'test', 'test'),
+                                (4, NULL, '', 'value'),
+                                (5, '', NULL, NULL)";
+                            await cmd.ExecuteNonQueryAsync();
+                        }
+
+                        // Read data asynchronously - this should not throw ArgumentNullException
+                        using (SqlCommand cmd = connection.CreateCommand())
+                        {
+                            cmd.CommandText = $"SELECT Id, VarcharMaxCol, NVarcharMaxCol, TextCol FROM [{tableName}] ORDER BY Id";
+                            using (SqlDataReader reader = await cmd.ExecuteReaderAsync())
+                            {
+                                // Row 1: All NULL values
+                                Assert.True(await reader.ReadAsync());
+                                Assert.Equal(1, reader.GetInt32(0));
+                                Assert.True(await reader.IsDBNullAsync(1));
+                                Assert.True(await reader.IsDBNullAsync(2));
+                                Assert.True(await reader.IsDBNullAsync(3));
+
+                                // Row 2: All empty strings
+                                Assert.True(await reader.ReadAsync());
+                                Assert.Equal(2, reader.GetInt32(0));
+                                Assert.Equal(string.Empty, await reader.GetFieldValueAsync<string>(1));
+                                Assert.Equal(string.Empty, await reader.GetFieldValueAsync<string>(2));
+                                Assert.Equal(string.Empty, await reader.GetFieldValueAsync<string>(3));
+
+                                // Row 3: Non-empty values
+                                Assert.True(await reader.ReadAsync());
+                                Assert.Equal(3, reader.GetInt32(0));
+                                Assert.Equal("test", await reader.GetFieldValueAsync<string>(1));
+                                Assert.Equal("test", await reader.GetFieldValueAsync<string>(2));
+                                Assert.Equal("test", await reader.GetFieldValueAsync<string>(3));
+
+                                // Row 4: Mixed NULL and values
+                                Assert.True(await reader.ReadAsync());
+                                Assert.Equal(4, reader.GetInt32(0));
+                                Assert.True(await reader.IsDBNullAsync(1));
+                                Assert.Equal(string.Empty, await reader.GetFieldValueAsync<string>(2));
+                                Assert.Equal("value", await reader.GetFieldValueAsync<string>(3));
+
+                                // Row 5: Mixed empty and NULL
+                                Assert.True(await reader.ReadAsync());
+                                Assert.Equal(5, reader.GetInt32(0));
+                                Assert.Equal(string.Empty, await reader.GetFieldValueAsync<string>(1));
+                                Assert.True(await reader.IsDBNullAsync(2));
+                                Assert.True(await reader.IsDBNullAsync(3));
+
+                                Assert.False(await reader.ReadAsync());
+                            }
+                        }
+                    }
+                }
+            }
+            finally
+            {
+                // Clean up test table
+                try
+                {
+                    using (SqlConnection connection = new SqlConnection(DataTestUtility.TCPConnectionString))
+                    {
+                        connection.Open();
+                        DataTestUtility.DropTable(connection, tableName);
+                    }
+                }
+                catch
+                {
+                    // Ignore cleanup errors
+                }
+            }
+        }
     }
 }