El mes pasado, cubrí un acertijo que consistía en hacer coincidir cada fila de una tabla con la coincidencia más cercana de otra tabla. Obtuve este acertijo de Karen Ly, una analista junior de renta fija en RBC. Cubrí dos soluciones relacionales principales que combinaban el operador APPLY con subconsultas basadas en TOP. La solución 1 siempre tuvo una escala cuadrática. La solución 2 funcionó bastante bien cuando se le proporcionó buenos índices de apoyo, pero sin esos índices también tuvo una escala cuádrica. En este artículo, cubro las soluciones iterativas que, a pesar de que los profesionales de SQL suelen desaprobarlas, proporcionan un escalado mucho mejor en nuestro caso, incluso sin una indexación óptima.
El desafío
Como recordatorio rápido, nuestro desafío involucra tablas llamadas T1 y T2, que creas con el siguiente código:
SET NOCOUNT ON;
IF DB_ID('testdb') IS NULL
CREATE DATABASE testdb;
GO
USE testdb;
DROP TABLE IF EXISTS dbo.T1, dbo.T2;
CREATE TABLE dbo.T1
(
keycol INT NOT NULL IDENTITY
CONSTRAINT PK_T1 PRIMARY KEY,
val INT NOT NULL,
othercols BINARY(100) NOT NULL
CONSTRAINT DFT_T1_col1 DEFAULT(0xAA)
);
CREATE TABLE dbo.T2
(
keycol INT NOT NULL IDENTITY
CONSTRAINT PK_T2 PRIMARY KEY,
val INT NOT NULL,
othercols BINARY(100) NOT NULL
CONSTRAINT DFT_T2_col1 DEFAULT(0xBB)
); Luego usa el siguiente código para completar las tablas con pequeños conjuntos de datos de muestra para verificar la corrección de sus soluciones:
TRUNCATE TABLE dbo.T1;
TRUNCATE TABLE dbo.T2;
INSERT INTO dbo.T1 (val)
VALUES(1),(1),(3),(3),(5),(8),(13),(16),(18),(20),(21);
INSERT INTO dbo.T2 (val)
VALUES(2),(2),(7),(3),(3),(11),(11),(13),(17),(19); Recuerde que el desafío era hacer coincidir cada fila de T1 con la fila de T2 donde la diferencia absoluta entre T2.val y T1.val es la más baja. En caso de empate, se supone que debe usar val ascendente, keycol orden ascendente como criterio de desempate.
Aquí está el resultado deseado para los datos de muestra dados:
keycol1 val1 othercols1 keycol2 val2 othercols2 ----------- ----------- ---------- ----------- ----------- ---------- 1 1 0xAA 1 2 0xBB 2 1 0xAA 1 2 0xBB 3 3 0xAA 4 3 0xBB 4 3 0xAA 4 3 0xBB 5 5 0xAA 4 3 0xBB 6 8 0xAA 3 7 0xBB 7 13 0xAA 8 13 0xBB 8 16 0xAA 9 17 0xBB 9 18 0xAA 9 17 0xBB 10 20 0xAA 10 19 0xBB 11 21 0xAA 10 19 0xBB
Para verificar el rendimiento de sus soluciones, necesita conjuntos más grandes de datos de muestra. Primero crea la función auxiliar GetNums, que genera una secuencia de enteros en un rango solicitado, utilizando el siguiente código:
DROP FUNCTION IF EXISTS dbo.GetNums;
GO
CREATE OR ALTER FUNCTION dbo.GetNums(@low AS BIGINT, @high AS BIGINT) RETURNS TABLE
AS
RETURN
WITH
L0 AS (SELECT c FROM (SELECT 1 UNION ALL SELECT 1) AS D(c)),
L1 AS (SELECT 1 AS c FROM L0 AS A CROSS JOIN L0 AS B),
L2 AS (SELECT 1 AS c FROM L1 AS A CROSS JOIN L1 AS B),
L3 AS (SELECT 1 AS c FROM L2 AS A CROSS JOIN L2 AS B),
L4 AS (SELECT 1 AS c FROM L3 AS A CROSS JOIN L3 AS B),
L5 AS (SELECT 1 AS c FROM L4 AS A CROSS JOIN L4 AS B),
Nums AS (SELECT ROW_NUMBER() OVER(ORDER BY (SELECT NULL)) AS rownum
FROM L5)
SELECT TOP(@high - @low + 1) @low + rownum - 1 AS n
FROM Nums
ORDER BY rownum;
GO Luego, completa T1 y T2 con el siguiente código, ajustando los parámetros que indican el número de filas y los valores máximos según sus necesidades:
DECLARE
@numrowsT1 AS INT = 1000000,
@maxvalT1 AS INT = 10000000,
@numrowsT2 AS INT = 1000000,
@maxvalT2 AS INT = 10000000;
TRUNCATE TABLE dbo.T1;
TRUNCATE TABLE dbo.T2;
INSERT INTO dbo.T1 WITH(TABLOCK) (val)
SELECT ABS(CHECKSUM(NEWID())) % @maxvalT1 + 1 AS val
FROM dbo.GetNums(1, @numrowsT1) AS Nums;
INSERT INTO dbo.T2 WITH(TABLOCK) (val)
SELECT ABS(CHECKSUM(NEWID())) % @maxvalT2 + 1 AS val
FROM dbo.GetNums(1, @numrowsT2) AS Nums; En este ejemplo, está completando las tablas con 1 000 000 de filas cada una, con valores en el rango de 1 a 10 000 000 en la columna val (densidad baja).
Solución 3, usando un cursor y una variable de tabla basada en disco
Una solución iterativa eficiente para nuestro desafío de coincidencia más cercana se basa en un algoritmo similar al algoritmo Merge join. La idea es aplicar un solo pase ordenado contra cada mesa usando cursores, evaluando los elementos de ordenación y desempate en cada ronda para decidir en qué lado avanzar, y emparejando las filas en el camino.
El paso ordenado contra cada tabla sin duda se beneficiará de los índices de soporte, pero la implicación de no tenerlos es que se llevará a cabo una clasificación explícita. Esto significa que la parte de clasificación incurrirá en una escala log n, pero eso es mucho menos severo que la escala cuadrática que obtiene de la Solución 2 en circunstancias similares.
Además, el rendimiento de las soluciones 1 y 2 se vio afectado por la densidad de la columna val. Con mayor densidad, el plan aplicó menos reenlaces. Por el contrario, dado que las soluciones iterativas realizan solo una pasada en cada una de las entradas, la densidad de la columna val no es un factor que afecte el rendimiento.
Utilice el siguiente código para crear índices de apoyo:
CREATE INDEX idx_val_key ON dbo.T1(val, keycol) INCLUDE(othercols); CREATE INDEX idx_val_key ON dbo.T2(val, keycol) INCLUDE(othercols);
Asegúrese de probar las soluciones con y sin estos índices.
Aquí está el código completo para la Solución 3:
SET NOCOUNT ON;
BEGIN TRAN;
DECLARE
@keycol1 AS INT, @val1 AS INT, @othercols1 AS BINARY(100),
@keycol2 AS INT, @val2 AS INT, @othercols2 AS BINARY(100),
@prevkeycol2 AS INT, @prevval2 AS INT, @prevothercols2 AS BINARY(100),
@C1 AS CURSOR, @C2 AS CURSOR,
@C1fetch_status AS INT, @C2fetch_status AS INT;
DECLARE @Result AS TABLE
(
keycol1 INT NOT NULL PRIMARY KEY,
val1 INT NOT NULL,
othercols1 BINARY(100) NOT NULL,
keycol2 INT NULL,
val2 INT NULL,
othercols2 BINARY(100) NULL
);
SET @C1 = CURSOR FORWARD_ONLY STATIC READ_ONLY FOR
SELECT keycol, val, othercols FROM dbo.T1 ORDER BY val, keycol;
SET @C2 = CURSOR FORWARD_ONLY STATIC READ_ONLY FOR
SELECT keycol, val, othercols FROM dbo.T2 ORDER BY val, keycol;
OPEN @C1;
OPEN @C2;
FETCH NEXT FROM @C2 INTO @keycol2, @val2, @othercols2;
SET @C2fetch_status = @@fetch_status;
SELECT @prevkeycol2 = @keycol2, @prevval2 = @val2, @prevothercols2 = @othercols2;
FETCH NEXT FROM @C1 INTO @keycol1, @val1, @othercols1;
SET @C1fetch_status = @@fetch_status;
WHILE @C1fetch_status = 0
BEGIN
IF @val1 <= @val2 OR @C2fetch_status <> 0
BEGIN
IF ABS(@val1 - @val2) < ABS(@val1 - @prevval2)
INSERT INTO @Result(keycol1, val1, othercols1, keycol2, val2, othercols2)
VALUES(@keycol1, @val1, @othercols1, @keycol2, @val2, @othercols2);
ELSE
INSERT INTO @Result(keycol1, val1, othercols1, keycol2, val2, othercols2)
VALUES(@keycol1, @val1, @othercols1, @prevkeycol2, @prevval2, @prevothercols2);
FETCH NEXT FROM @C1 INTO @keycol1, @val1, @othercols1;
SET @C1fetch_status = @@fetch_status;
END
ELSE IF @C2fetch_status = 0
BEGIN
IF @val2 > @prevval2
SELECT @prevkeycol2 = @keycol2, @prevval2 = @val2, @prevothercols2 = @othercols2;
FETCH NEXT FROM @C2 INTO @keycol2, @val2, @othercols2;
SET @C2fetch_status = @@fetch_status;
END;
END;
SELECT
keycol1, val1, SUBSTRING(othercols1, 1, 1) AS othercols1,
keycol2, val2, SUBSTRING(othercols2, 1, 1) AS othercols2
FROM @Result;
COMMIT TRAN; El código usa una variable de tabla llamada @Result para almacenar las coincidencias y finalmente las devuelve consultando la variable de tabla. Tenga en cuenta que el código realiza el trabajo en una transacción para reducir el registro.
El código usa variables de cursor llamadas @C1 y @C2 para iterar a través de las filas en T1 y T2, respectivamente, en ambos casos ordenados por val, keycol. Las variables locales se utilizan para almacenar los valores de fila actuales de cada cursor (@keycol1, @val1 y @othercols1 para @C1, y @keycol2, @val2 y @othercols2 para @C2). Las variables locales adicionales almacenan los valores de fila anteriores de @C2 (@prevkeycol2, @prevval2 y @prevothercols2). Las variables @C1fetch_status y @C2fetch_status mantienen el estado de la última búsqueda del cursor respectivo.
Después de declarar y abrir ambos cursores, el código obtiene una fila de cada cursor en las respectivas variables locales e inicialmente almacena los valores de fila actuales de @C2 también en las variables de fila anteriores. Luego, el código ingresa en un bucle que sigue ejecutándose mientras la última recuperación de @C1 fue exitosa (@C1fetch_status =0). El cuerpo del bucle aplica el siguiente pseudocódigo en cada ronda:
If @val1 <= @val2 or reached end of @C2
Begin
If absolute difference between @val1 and @val2 is less than between @val1 and @prevval2
Add row to @Result with current row values from @C1 and current row values from @C2
Else
Add row to @Result with current row values from @C1 and previous row values from @C2
Fetch next row from @C1
End
Else if last fetch from @C2 was successful
Begin
If @val2 > @prevval2
Set variables holding @C2’s previous row values to values of current row variables
Fetch next row from @C2
End Luego, el código simplemente consulta la variable de tabla @Result para devolver todas las coincidencias.
Utilizando los grandes conjuntos de datos de muestra (1 000 000 de filas en cada tabla), con una indexación óptima, esta solución tardó 38 segundos en completarse en mi sistema y realizó 28 240 lecturas lógicas. Por supuesto, la escala de esta solución es entonces lineal. Sin una indexación óptima, tardó 40 segundos en completarse (¡solo 2 segundos más!) y realizó 29 519 lecturas lógicas. La parte de clasificación en esta solución tiene escala n log n.
Solución 4, usando un cursor y una variable de tabla optimizada para memoria
En un intento por mejorar el rendimiento del enfoque iterativo, una cosa que podría intentar es reemplazar el uso de la variable de tabla basada en disco con una optimizada para memoria. Dado que la solución implica escribir 1 000 000 de filas en la variable de la tabla, esto podría resultar en una mejora no despreciable.
En primer lugar, debe habilitar In-Memory OLTP en la base de datos mediante la creación de un grupo de archivos marcado como CONTIENE MEMORIA_OPTIMIZADA_DATOS, y dentro de él un contenedor que apunte a una carpeta en el sistema de archivos. Suponiendo que creó una carpeta principal llamada C:\IMOLTP\, use el siguiente código para aplicar estos dos pasos:
ALTER DATABASE testdb
ADD FILEGROUP testdb_MO CONTAINS MEMORY_OPTIMIZED_DATA;
ALTER DATABASE testdb
ADD FILE ( NAME = testdb_dir,
FILENAME = 'C:\IMOLTP\testdb_dir' )
TO FILEGROUP testdb_MO; El siguiente paso es crear un tipo de tabla optimizada para memoria como plantilla para nuestra variable de tabla ejecutando el siguiente código:
DROP TYPE IF EXISTS dbo.TYPE_closestmatch;
GO
CREATE TYPE dbo.TYPE_closestmatch AS TABLE
(
keycol1 INT NOT NULL PRIMARY KEY NONCLUSTERED,
val1 INT NOT NULL,
othercols1 BINARY(100) NOT NULL,
keycol2 INT NULL,
val2 INT NULL,
othercols2 BINARY(100) NULL
)
WITH (MEMORY_OPTIMIZED = ON); Luego, en lugar de la declaración original de la variable de tabla @Result, usaría el siguiente código:
DECLARE @Result AS dbo.TYPE_closestmatch;
Aquí está el código completo de la solución:
SET NOCOUNT ON;
USE testdb;
BEGIN TRAN;
DECLARE
@keycol1 AS INT, @val1 AS INT, @othercols1 AS BINARY(100),
@keycol2 AS INT, @val2 AS INT, @othercols2 AS BINARY(100),
@prevkeycol2 AS INT, @prevval2 AS INT, @prevothercols2 AS BINARY(100),
@C1 AS CURSOR, @C2 AS CURSOR,
@C1fetch_status AS INT, @C2fetch_status AS INT;
DECLARE @Result AS dbo.TYPE_closestmatch;
SET @C1 = CURSOR FORWARD_ONLY STATIC READ_ONLY FOR
SELECT keycol, val, othercols FROM dbo.T1 ORDER BY val, keycol;
SET @C2 = CURSOR FORWARD_ONLY STATIC READ_ONLY FOR
SELECT keycol, val, othercols FROM dbo.T2 ORDER BY val, keycol;
OPEN @C1;
OPEN @C2;
FETCH NEXT FROM @C2 INTO @keycol2, @val2, @othercols2;
SET @C2fetch_status = @@fetch_status;
SELECT @prevkeycol2 = @keycol2, @prevval2 = @val2, @prevothercols2 = @othercols2;
FETCH NEXT FROM @C1 INTO @keycol1, @val1, @othercols1;
SET @C1fetch_status = @@fetch_status;
WHILE @C1fetch_status = 0
BEGIN
IF @val1 <= @val2 OR @C2fetch_status <> 0
BEGIN
IF ABS(@val1 - @val2) < ABS(@val1 - @prevval2)
INSERT INTO @Result(keycol1, val1, othercols1, keycol2, val2, othercols2)
VALUES(@keycol1, @val1, @othercols1, @keycol2, @val2, @othercols2);
ELSE
INSERT INTO @Result(keycol1, val1, othercols1, keycol2, val2, othercols2)
VALUES(@keycol1, @val1, @othercols1, @prevkeycol2, @prevval2, @prevothercols2);
FETCH NEXT FROM @C1 INTO @keycol1, @val1, @othercols1;
SET @C1fetch_status = @@fetch_status;
END
ELSE IF @C2fetch_status = 0
BEGIN
IF @val2 > @prevval2
SELECT @prevkeycol2 = @keycol2, @prevval2 = @val2, @prevothercols2 = @othercols2;
FETCH NEXT FROM @C2 INTO @keycol2, @val2, @othercols2;
SET @C2fetch_status = @@fetch_status;
END;
END;
SELECT
keycol1, val1, SUBSTRING(othercols1, 1, 1) AS othercols1,
keycol2, val2, SUBSTRING(othercols2, 1, 1) AS othercols2
FROM @Result;
COMMIT TRAN; Con la indexación óptima implementada, esta solución tardó 27 segundos en completarse en mi máquina (en comparación con los 38 segundos con la variable de tabla basada en disco), y sin la indexación óptima tardó 29 segundos en completarse (en comparación con 40 segundos). Eso es una reducción cercana al 30 por ciento en el tiempo de ejecución.
Solución 5, usando SQL CLR
Otra forma de mejorar aún más el rendimiento del enfoque iterativo es implementar la solución mediante SQL CLR, dado que la mayor parte de la sobrecarga de la solución T-SQL se debe a las ineficiencias de la obtención y el bucle del cursor en T-SQL.
Aquí está el código completo de la solución que implementa el mismo algoritmo que usé en las Soluciones 3 y 4 con C#, usando objetos SqlDataReader en lugar de cursores T-SQL:
using System;
using System.Data;
using System.Data.SqlClient;
using System.Data.SqlTypes;
using Microsoft.SqlServer.Server;
public partial class ClosestMatch
{
[SqlProcedure]
public static void GetClosestMatches()
{
using (SqlConnection conn = new SqlConnection("data source=MyServer\\MyInstance;Database=testdb;Trusted_Connection=True;MultipleActiveResultSets=true;"))
{
SqlCommand comm1 = new SqlCommand();
SqlCommand comm2 = new SqlCommand();
comm1.Connection = conn;
comm2.Connection = conn;
comm1.CommandText = "SELECT keycol, val, othercols FROM dbo.T1 ORDER BY val, keycol;";
comm2.CommandText = "SELECT keycol, val, othercols FROM dbo.T2 ORDER BY val, keycol;";
SqlMetaData[] columns = new SqlMetaData[6];
columns[0] = new SqlMetaData("keycol1", SqlDbType.Int);
columns[1] = new SqlMetaData("val1", SqlDbType.Int);
columns[2] = new SqlMetaData("othercols1", SqlDbType.Binary, 100);
columns[3] = new SqlMetaData("keycol2", SqlDbType.Int);
columns[4] = new SqlMetaData("val2", SqlDbType.Int);
columns[5] = new SqlMetaData("othercols2", SqlDbType.Binary, 100);
SqlDataRecord record = new SqlDataRecord(columns);
SqlContext.Pipe.SendResultsStart(record);
conn.Open();
SqlDataReader reader1 = comm1.ExecuteReader();
SqlDataReader reader2 = comm2.ExecuteReader();
SqlInt32 keycol1 = SqlInt32.Null;
SqlInt32 val1 = SqlInt32.Null;
SqlBinary othercols1 = SqlBinary.Null;
SqlInt32 keycol2 = SqlInt32.Null;
SqlInt32 val2 = SqlInt32.Null;
SqlBinary othercols2 = SqlBinary.Null;
SqlInt32 prevkeycol2 = SqlInt32.Null;
SqlInt32 prevval2 = SqlInt32.Null;
SqlBinary prevothercols2 = SqlBinary.Null;
Boolean reader2foundrow = reader2.Read();
if (reader2foundrow)
{
keycol2 = reader2.GetSqlInt32(0);
val2 = reader2.GetSqlInt32(1);
othercols2 = reader2.GetSqlBinary(2);
prevkeycol2 = keycol2;
prevval2 = val2;
prevothercols2 = othercols2;
}
Boolean reader1foundrow = reader1.Read();
if (reader1foundrow)
{
keycol1 = reader1.GetSqlInt32(0);
val1 = reader1.GetSqlInt32(1);
othercols1 = reader1.GetSqlBinary(2);
}
while (reader1foundrow)
{
if (val1 <= val2 || !reader2foundrow)
{
if (Math.Abs((int)(val1 - val2)) < Math.Abs((int)(val1 - prevval2)))
{
record.SetSqlInt32(0, keycol1);
record.SetSqlInt32(1, val1);
record.SetSqlBinary(2, othercols1);
record.SetSqlInt32(3, keycol2);
record.SetSqlInt32(4, val2);
record.SetSqlBinary(5, othercols2);
SqlContext.Pipe.SendResultsRow(record);
}
else
{
record.SetSqlInt32(0, keycol1);
record.SetSqlInt32(1, val1);
record.SetSqlBinary(2, othercols1);
record.SetSqlInt32(3, prevkeycol2);
record.SetSqlInt32(4, prevval2);
record.SetSqlBinary(5, prevothercols2);
SqlContext.Pipe.SendResultsRow(record);
}
reader1foundrow = reader1.Read();
if (reader1foundrow)
{
keycol1 = reader1.GetSqlInt32(0);
val1 = reader1.GetSqlInt32(1);
othercols1 = reader1.GetSqlBinary(2);
}
}
else if (reader2foundrow)
{
if (val2 > prevval2)
{
prevkeycol2 = keycol2;
prevval2 = val2;
prevothercols2 = othercols2;
}
reader2foundrow = reader2.Read();
if (reader2foundrow)
{
keycol2 = reader2.GetSqlInt32(0);
val2 = reader2.GetSqlInt32(1);
othercols2 = reader2.GetSqlBinary(2);
}
}
}
SqlContext.Pipe.SendResultsEnd();
}
}
} Para conectarse a la base de datos, normalmente usaría la opción "conexión de contexto =verdadero" en lugar de una cadena de conexión completa. Lamentablemente, esta opción no está disponible cuando necesita trabajar con varios conjuntos de resultados activos. Nuestra solución emula el trabajo paralelo con dos cursores usando dos objetos SqlDataReader y, por lo tanto, necesita una cadena de conexión completa, con la opción MultipleActiveResultSets=true. Aquí está la cadena de conexión completa:
"data source=MyServer\\MyInstance;Database=testdb;Trusted_Connection=True;MultipleActiveResultSets=true;"
Por supuesto, en su caso necesitaría reemplazar MyServer\\MyInstance con los nombres de su servidor e instancia (si corresponde).
Además, el hecho de que no haya utilizado "context connection=true" en lugar de una cadena de conexión explícita significa que el ensamblado necesita acceso a un recurso externo y, por lo tanto, es de confianza. Normalmente, lograría esto al firmarlo con un certificado o una clave asimétrica que tenga un inicio de sesión correspondiente con el permiso correcto, o incluirlo en la lista blanca usando el procedimiento sp_add_trusted_assembly. En aras de la simplicidad, estableceré la opción de base de datos TRUSTWORTHY en ON y especificaré el conjunto de permisos EXTERNAL_ACCESS al crear el ensamblaje. El siguiente código implementa la solución en la base de datos:
EXEC sys.sp_configure 'advanced', 1;
RECONFIGURE;
EXEC sys.sp_configure 'clr enabled', 1;
EXEC sys.sp_configure 'clr strict security', 0;
RECONFIGURE;
EXEC sys.sp_configure 'advanced', 0;
RECONFIGURE;
ALTER DATABASE testdb SET TRUSTWORTHY ON;
USE testdb;
DROP PROC IF EXISTS dbo.GetClosestMatches;
DROP ASSEMBLY IF EXISTS ClosestMatch;
CREATE ASSEMBLY ClosestMatch
FROM 'C:\ClosestMatch\ClosestMatch\bin\Debug\ClosestMatch.dll'
WITH PERMISSION_SET = EXTERNAL_ACCESS;
GO
CREATE PROCEDURE dbo.GetClosestMatches
AS EXTERNAL NAME ClosestMatch.ClosestMatch.GetClosestMatches; El código habilita CLR en la instancia, deshabilita la opción de seguridad estricta de CLR, establece la opción de base de datos TRUSTWORTHY en ON, crea el ensamblado y crea el procedimiento GetClosestMatches.
Use el siguiente código para probar el procedimiento almacenado:
EXEC dbo.GetClosestMatches;
La solución CLR tardó 8 segundos en completarse en mi sistema con indexación óptima y 9 segundos sin ella. Esa es una mejora de rendimiento bastante impresionante en comparación con todas las demás soluciones, tanto relacionales como iterativas.
Conclusión
Las soluciones iterativas suelen estar mal vistas en la comunidad de SQL, ya que no siguen el modelo relacional. Sin embargo, la realidad es que a veces no puede crear soluciones relacionales de buen rendimiento y el rendimiento es una prioridad. Con un enfoque iterativo, no está limitado a los algoritmos a los que tiene acceso el optimizador de SQL Server, sino que puede implementar cualquier algoritmo que desee. Como se demostró en este artículo, al usar un algoritmo de combinación, pudo lograr la tarea con un solo paso ordenado en cada una de las entradas. Usando cursores T-SQL y una variable de tabla basada en disco, obtuvo un rendimiento y una escalabilidad razonables. Pudo mejorar el rendimiento en aproximadamente un 30 por ciento al cambiar a una variable de tabla optimizada para memoria, y mucho más al usar SQL CLR.
