Acelerar uniones de múltiples tablas con filtros en tiempo de ejecución - Hologres

En las consultas de unión de múltiples tablas, los datos que no cumplen la condición de unión incrementan la sobrecarga de E/S y degradan el rendimiento de la consulta. En escenarios de unión de múltiples tablas, los filtros en tiempo de ejecución generan automáticamente filtros ligeros que descartan los datos no coincidentes durante la fase de escaneo. Esto reduce la sobrecarga de E/S y mejora el rendimiento de la consulta. Esta función es compatible con Hash Join desde Hologres V2.0 y se extendió a escenarios de Cross Join en la V4.2.

Contexto

Escenarios de uso

Hologres admite filtros en tiempo de ejecución desde la versión V2.0. Esta función se utiliza comúnmente en escenarios de Hash Join que involucran dos o más tablas, especialmente cuando una tabla grande se une a una pequeña. No se requiere configuración manual. El optimizador y el motor de ejecución optimizan automáticamente el comportamiento de filtrado de la unión al momento de la consulta, lo que reduce la sobrecarga de E/S y mejora el rendimiento de la unión.

A partir de la V4.2, la capacidad de filtro en tiempo de ejecución se extiende a escenarios de Cross Join. Esto optimiza específicamente el patrón SQL común donde una subconsulta escalar calcula un resultado agregado y lo usa como condición de filtro en una tabla grande. Para obtener más información, consulte Cross Join support for runtime filters (New in V4.2).

Funcionamiento

Filtros en tiempo de ejecución para hash join

Cuando se unen dos tablas, los datos de una se cargan en una tabla hash y los datos de la otra se comparan con ella. El proceso de unión tiene dos lados:

Build side: El lado que construye la tabla hash, correspondiente al nodo Hash en el plan de ejecución.
Probe side: El lado que lee los datos y los compara con la tabla hash del build side.

Por lo general, la tabla pequeña sirve como build side y la tabla grande como probe side.

Los filtros en tiempo de ejecución funcionan mediante la construcción de un filtro ligero a partir de la distribución de datos del build side y su envío al probe side para descartar datos. Esto reduce la cantidad de datos procesados del probe side en el Hash Join y minimiza el tráfico de red, mejorando así el rendimiento de la unión. En consecuencia, esta función es más efectiva para uniones entre tablas grandes y pequeñas con una disparidad de tamaño significativa, ya que ofrece mayores mejoras de rendimiento que una unión estándar.

Filtros en tiempo de ejecución para cross join (Novedad en V4.2)

Antes de la V4.2, los filtros en tiempo de ejecución solo cubrían Hash Join. Sin embargo, los usuarios suelen utilizar subconsultas escalares para calcular valores agregados, como min o max, y luego usar el resultado como condición de filtro en una tabla grande. En el plan de ejecución, este patrón SQL produce un Cross Join:

Build side: El resultado de la subconsulta escalar, que es exactamente una fila.
Probe side: El escaneo de la tabla grande.

La V4.2 introduce un nuevo tipo de filtro, ScalarFilter. El optimizador identifica automáticamente este patrón y envía el valor escalar evaluado desde el build side hacia el ScanNode del probe side. Esto permite el filtrado a nivel de fila y de RowGroup durante la fase de escaneo, evitando un escaneo completo de la tabla seguido de comparaciones fila por fila.

Límites y condiciones de activación

Límites

Los filtros en tiempo de ejecución solo son compatibles con Hologres V2.0 y versiones posteriores.
En escenarios de Hash Join, la V2.0 admite filtros en tiempo de ejecución solo cuando la condición de unión contiene un único campo. A partir de la V2.1, los filtros en tiempo de ejecución admiten varios campos.
El filtro en tiempo de ejecución TopN solo es compatible con la V4.0 y versiones posteriores, y se utiliza para mejorar el rendimiento en cálculos TopN de una sola tabla.
El filtro en tiempo de ejecución Cross Join (ScalarFilter) solo es compatible con la V4.2 y versiones posteriores.

Condiciones de activación

Escenarios de Hash Join

El motor activa un filtro en tiempo de ejecución automáticamente cuando se cumplen todas las siguientes condiciones:

El probe side tiene 100,000 filas o más.
La proporción de datos escaneados del build side respecto al probe side es de 0.1 o menos. Cuanto menor sea la proporción, mayor será la probabilidad de que se active el filtro.
La proporción de datos de salida de la unión respecto a los datos del probe side es de 0.1 o menos. Cuanto menor sea la proporción, mayor será la probabilidad de que se active el filtro.

Escenarios de Cross Join (V4.2+)

El optimizador genera automáticamente un ScalarFilter cuando se cumplen las siguientes condiciones:

El build side del Cross Join tiene un recuento de filas estadístico de 1 y su distribución es Replicated.
Los predicados de filtro en el probe side contienen condiciones de comparación con expresiones del build side, tales como >, <, >=, <= o BETWEEN.

Tipos de filtros en tiempo de ejecución

Los filtros en tiempo de ejecución se pueden clasificar según las siguientes dos dimensiones.

Por alcance de shuffle (para hash join)

Tipo	Versiones compatibles	Escenarios
Local	V2.0+	Se utiliza cuando los datos del probe side no necesitan redistribuirse. Un filtro en tiempo de ejecución Local puede usarse si las claves de unión del build side y del probe side comparten la misma distribución, si los datos del build side se transmiten al probe side, o si los datos del build side se redistribuyen para coincidir con la distribución del probe side. Este tipo solo reduce la cantidad de datos escaneados y los datos procesados por el Hash Join.
Global	V2.2+	Se utiliza cuando los datos del probe side necesitan redistribuirse. El filtro en tiempo de ejecución se aplica antes de la redistribución de datos, lo que reduce el tráfico de red.

Nota

No necesita especificar el tipo. El motor lo selecciona de forma adaptativa.

Por tipo de filtro

Tipo	Versiones compatibles	Descripción
Filtro Bloom	V2.0+	Este filtro es probabilístico y puede tener falsos positivos, lo que significa que algunos datos podrían no filtrarse. Sin embargo, tiene una amplia aplicabilidad y mantiene una alta eficiencia de filtrado incluso cuando el build side contiene una gran cantidad de datos.
Filtro In	V2.0+	Se utiliza cuando el build side tiene un NDV bajo (número de valores distintos). Construye un HashSet a partir de los datos del build side y lo envía al probe side para su filtrado. Filtra con precisión todos los datos requeridos y puede usarse junto con un índice de mapa de bits.
Filtro MinMax	V2.0+	Envía los valores mínimo y máximo de los datos del build side al probe side para su filtrado. Puede aprovechar los metadatos para omitir archivos completos o un lote de datos, reduciendo los costos de E/S.
ScalarFilter	V4.2+	Diseñado específicamente para escenarios de Cross Join. Cuando el build side tiene exactamente una fila, el motor envía el valor escalar al ScanNode del probe side para realizar el filtrado durante la fase de escaneo.

Nota

No necesita especificar el tipo de filtro. Hologres selecciona un tipo de forma adaptativa basándose en las condiciones de unión en tiempo de ejecución.

Compatibilidad de Cross Join con filtros en tiempo de ejecución (Novedad en V4.2)

Motivación

Los usuarios suelen calcular valores agregados con subconsultas escalares y usar los resultados como condiciones de filtro en una tabla grande. Antes de la V4.2, este patrón SQL presentaba los siguientes problemas en el plan de ejecución:

El ScanNode del probe side no podía aprovechar el valor del build side para un filtrado temprano y tenía que realizar un escaneo completo de la tabla.
Después del escaneo, el Cross Join realizaba una comparación fila por fila, lo que causaba un desperdicio significativo de E/S.
El filtro en tiempo de ejecución existente solo cubría Hash Join y no era compatible con Cross Join.

La V4.2 introduce el ScalarFilter para resolver este problema. Esta función está habilitada de forma predeterminada y no requiere ninguna acción por parte del usuario.

Comparación de SQL típico y plan de ejecución

SQL típico:

-- t1 is a large table; the aggregated min/max result from t2 is exactly 1 row.
SELECT * FROM t1
WHERE a >= (SELECT min(a) FROM t2 WHERE b BETWEEN 0 AND 1);

Antes de la optimización (sin filtro en tiempo de ejecución):

 Cross Join
   ->  Seq Scan on t1             -- Full table scan, no early filtering
   ->  Aggregate                  -- Build side min/max, 1-row result
         ->  Seq Scan on t2

El probe side debe escanear todos los datos y luego el Cross Join compara las filas una por una, lo que causa un desperdicio significativo de E/S.

Después de la optimización (V4.2 con ScalarFilter):

 Cross Join
   Runtime Filter Build Expr: (min(t2.a)), (max(t2.a))
   ->  Seq Scan on t1             -- Receives ScalarFilter, filters non-matching rows and RowGroups during scan
         Runtime Filter Target Expr: (t1.a >= ${1}) AND (t1.a <= ${2})
   ->  Aggregate
         ->  Seq Scan on t2

Una vez evaluado el build side, los valores reales se envían al ScanNode del probe side, lo que permite completar el filtrado durante la fase de escaneo.

Más escenarios típicos

-- Scenario 1: Single table + scalar subquery
SELECT * FROM t1
WHERE a >= (SELECT min(a) FROM t2 WHERE b BETWEEN 0 AND 1);
-- Scenario 2: Multi-table + scalar subquery
SELECT * FROM t1, t3
WHERE (SELECT min(a) FROM t2) <= t1.a
  AND t3.a <= (SELECT max(a) FROM t2);
-- Scenario 3: CTE + Cross Join
WITH r AS (SELECT MIN(a) AS lo, MAX(a) AS hi FROM t2)
SELECT t1.* FROM t1, r
WHERE t1.a >= r.lo AND t1.a <= r.hi;

Funcionamiento

Identificación del optimizador: Cuando el build side de un Cross Join tiene un recuento de filas estadístico de 1 y una distribución Replicated, el optimizador extrae expresiones de comparación de los predicados (expandiendo BETWEEN en >= y <=) y genera candidatos de ScalarFilter.
Envío en tiempo de ejecución: Durante la fase Open, el Cross Join consume los datos del build side, evalúa la expresión build_expr para producir un valor escalar, construye un ScalarFilter y lo publica en el ScanNode del probe side.
Filtrado temprano en ScanNode: Después de que el NiagaraScan del probe side recibe el ScalarFilter, reemplaza los marcadores de posición en la expresión target_expr con valores reales y genera dos niveles de filtrado:
- Filtrado a nivel de fila (evaluación conjuntiva).
- Filtrado a nivel de RowGroup (El motor de almacenamiento omite bloques de datos no coincidentes).

Límites

El build side debe tener exactamente una fila. El optimizador genera un ScalarFilter solo cuando el build side tiene un recuento de filas estadístico de 1 y una distribución Replicated. Si el build side no produce exactamente una fila en tiempo de ejecución, Hologres reporta un error RT_CHECK.
Si el valor del build side es NULL, Hologres publica un filtro FILTER_ALL (que filtra todas las filas), borra los datos del build side para cortocircuitar el Cross Join y devuelve un conjunto de resultados vacío.
Las columnas de diccionario no son compatibles. Si la columna del probe side referenciada por target_expr es una columna codificada por diccionario (tipo DICTIONARY), el ScalarFilter no se aplica.
Los tipos de datos compatibles incluyen INT8, UINT8, INT16, UINT16, INT32, UINT32, INT64, UINT64, DATE32, TIMESTAMP, FLOAT, DOUBLE y STRING. Si un tipo de dato no es compatible, el ScalarFilter no se envía y no se reporta ningún error.
Los operadores de comparación compatibles incluyen >, <, >= y <=. BETWEEN se expande en dos comparaciones de rango.

Parámetros GUC

Parámetro	Tipo	Valor predeterminado	Nivel	Descripción
`hg_experimental_generate_runtime_scalar_filter`	bool	`true`	`PGC_USERSET`	Controla si se genera un ScalarFilter para Cross Joins. Esta función está habilitada de forma predeterminada y no requiere ninguna acción por parte del usuario.

Para deshabilitar esta función, ejecute el siguiente comando:

SET hg_experimental_generate_runtime_scalar_filter = off;

Nuevos campos en la salida de EXPLAIN

Cuando ScalarFilter está habilitado, el nodo Cross Join muestra la siguiente información:

Runtime Filter Build Expr: La expresión del build side, como (min(t2.a)) y (max(t2.a)).
Runtime Filter Target Expr: La expresión de filtro objetivo en el probe side, como (t1.a >= ${1}) AND (t1.a <= ${2}). En esta expresión, ${N} es un marcador de posición que se reemplaza en tiempo de ejecución con el valor del build side correspondiente al filter_id.

Beneficios de rendimiento

En el conjunto de datos TPC-DS de 10 TB, el siguiente patrón SQL es común:

DELETE FROM inventory
WHERE inv_date_sk >= (SELECT min(d_date_sk) FROM date_dim WHERE d_date BETWEEN 'INV_S_1' AND 'INV_E_1')
  AND inv_date_sk <= (SELECT max(d_date_sk) FROM date_dim WHERE d_date BETWEEN 'INV_S_1' AND 'INV_E_1');

Con la optimización ScalarFilter de la V4.2, el tiempo de consulta disminuye de 2,672 segundos a 0,552 segundos, lo que representa un aumento de rendimiento de aproximadamente 4.8x.

Verificación de filtros en tiempo de ejecución

Los siguientes ejemplos demuestran cómo verificar los efectos de los filtros en tiempo de ejecución en diferentes escenarios.

Ejemplo 1: Condición de unión de columna única (tipo local)

BEGIN;
CREATE TABLE test1 (x int, y int);
CALL set_table_property('test1', 'distribution_key', 'x');
CREATE TABLE test2 (x int, y int);
CALL set_table_property('test2', 'distribution_key', 'x');
END;
INSERT INTO test1 SELECT t, t FROM generate_series(1, 100000) t;
INSERT INTO test2 SELECT t, t FROM generate_series(1, 1000) t;
ANALYZE test1;
ANALYZE test2;
EXPLAIN ANALYZE SELECT * FROM test1 JOIN test2 ON test1.x = test2.x;

Plan de ejecución:

QUERY PLAN
Gather  (cost=0.00..10.20 rows=1000 width=16)
[40:1 id=100002 dop=1 time=9/9/9ms rows=1000(1000/1000/1000) mem=16/16/16KB open=2/2/2ms get_next=7/7/7ms]
  -> Hash Join  (cost=0.00..10.16 rows=1000 width=16)
      Hash Cond: (test1.x = test2.x)
      Runtime Filter Cond: (test1.x = test2.x)
      [id=8 dop=40 time=9/4/2ms rows=1000(39/25/17) mem=6/5/5KB open=4/1/0ms get_next=6/2/0ms]
      -> Local Gather  (cost=0.00..5.11 rows=1000000 width=8)
          [id=3 dop=40 time=6/1/0ms rows=1000(39/25/17) mem=600/600/600B open=1/0/0ms get_next=6/1/0ms local_dop=1/1/1]
          -> Seq Scan on test1  (cost=0.00..5.10 rows=1000000 width=8)
              Runtime Filter Target Expr: test1.x
              [id=2 split_count=40 time=11/7/7ms rows=1000(39/25/17) mem=41/41/41KB open=11/7/7ms get_next=0/0/0ms scan_rows=1000000 25270/25000/24697))]
      -> Hash  (cost=5.00..5.00 rows=1000 width=8)
          [id=7 dop=40 time=3/1/0ms rows=1000(39/25/17) mem=396/396/396KB open=3/1/0ms get_next=1/0/0ms rehash=1/1/1 hash_mem=384/384/384KB]
          -> Local Gather  (cost=0.00..5.00 rows=1000 width=8)
              [id=5 dop=40 time=1/0/0ms rows=1000(39/25/17) mem=0/0/0B open=1/0/0ms get_next=1/0/0ms local_dop=0/0/0]
              -> Seq Scan on test2  (cost=0.00..5.00 rows=1000 width=8)
                  [id=4 split_count=40 time=1/0/0ms rows=1000(39/25/17) mem=528/528/528B open=1/0/0ms get_next=1/0/0ms scan_rows=1000(39/25/17)]

La tabla test2 tiene 1,000 filas y la tabla test1 tiene 100,000 filas. La proporción de tamaño de datos entre build side y probe side es de 0.01 (menos de 0.1), lo que cumple la condición de activación predeterminada para filtros en tiempo de ejecución.
El escaneo del probe side en test1 muestra Runtime Filter Target Expr, lo que indica que se envió un filtro en tiempo de ejecución.
En el probe side, scan_rows es 100,000 (el número de filas leídas del almacenamiento), mientras que rows es 1,000 (el número de filas después del filtrado). Esta diferencia demuestra el efecto de filtrado.

Ejemplo 2: Condición de unión de múltiples columnas (V2.1+, tipo local)

DROP TABLE IF EXISTS test1, test2;
BEGIN;
CREATE TABLE test1 (x int, y int);
CREATE TABLE test2 (x int, y int);
END;
INSERT INTO test1 SELECT t, t FROM generate_series(1, 1000000) t;
INSERT INTO test2 SELECT t, t FROM generate_series(1, 1000) t;
ANALYZE test1;
ANALYZE test2;
EXPLAIN ANALYZE SELECT * FROM test1 JOIN test2 ON test1.x = test2.x AND test1.y = test2.y;

Plan de ejecución:

QUERY PLAN
Gather  (cost=0.00..10.46 rows=1000 width=16)
[40:1 id=100003 dop=1 time=6/6/6ms rows=1000(1000/1000/1000) mem=600/600/600B open=0/0/0ms get_next=6/6/6ms]
  -> Hash Join  (cost=0.00..10.43 rows=1000 width=16)
       Hash Cond: ((test1.x = test2.x) AND (test1.y = test2.y))
       Runtime Filter Cond: ((test1.x = test2.x) AND (test1.y = test2.y))
       [id=8 dop=40 time=5/3/3ms rows=1000(1000/25/0) mem=40/2/1KB open=1/0/0ms get_next=4/3/3ms]
       -> Local Gather  (cost=0.00..5.11 rows=1000000 width=8)
            [id=5 dop=40 time=4/3/3ms rows=1000(1000/25/0) mem=600/600/600B open=0/0/0ms get_next=4/3/3ms local_dop=1/1/1]
            -> Seq Scan on test1  (cost=0.00..5.10 rows=1000000 width=8)
                 Runtime Filter Target Expr: (test1.x AND test1.y)
                 [id=4 split_count=40 time=7/5/5ms rows=1000(1000/25/0) mem=49/11/9KB open=7/5/5ms get_next=1/0/0ms scan_rows=1000000(32768/25000/24576)]
       -> Hash  (cost=5.02..5.02 rows=40000 width=8)
            [id=7 dop=40 time=1/0/0ms rows=40000(1000/1000/1000) mem=417/417/417KB open=1/0/0ms get_next=1/0/0ms rehash=1/1/1 hash_mem=384/384/384KB]
            -> Broadcast  (cost=0.00..5.02 rows=40000 width=8)
                 [40:40 id=100002 dop=40 time=1/0/0ms rows=40000(1000/1000/1000) mem=0/0/0B open=1/0/0ms get_next=0/0/0ms * ]
                 -> Local Gather  (cost=0.00..5.00 rows=1000 width=8)
                      [id=3 dop=40 time=1/0/0ms rows=1000(1000/25/0) mem=600/600/600B open=1/0/0ms get_next=0/0/0ms local_dop=1/1/1]
                      -> Seq Scan on test2  (cost=0.00..5.00 rows=1000 width=8)
                           [id=2 split_count=40 time=1/0/0ms rows=1000(1000/25/0) mem=528/528/528B open=0/0/0ms get_next=1/0/0ms scan_rows=1000(1000/1000/1000)]

La condición de unión contiene varias columnas y el filtro en tiempo de ejecución también se genera para varias columnas.
Los datos del build side se transmiten, por lo que se utiliza un filtro en tiempo de ejecución Local.

Ejemplo 3: Tipo Global (V2.2+, unión con redistribución)

SET hg_experimental_enable_result_cache = OFF;
DROP TABLE IF EXISTS test1, test2;
BEGIN;
CREATE TABLE test1 (x int, y int);
CREATE TABLE test2 (x int, y int);
END;
INSERT INTO test1 SELECT t, t FROM generate_series(1, 100000) t;
INSERT INTO test2 SELECT t, t FROM generate_series(1, 1000) t;
ANALYZE test1;
ANALYZE test2;
EXPLAIN ANALYZE SELECT * FROM test1 JOIN test2 ON test1.x = test2.x;

Plan de ejecución:

QUERY PLAN
   -> Hash Join  (cost=0.00..10.08 rows=1000 width=16)
      Hash Cond: (test1.x = test2.x)
      Runtime Filter Cond: (test1.x = test2.x)
      [id=9 dop=40 time=10/8/8ms rows=1000(34/25/13) mem=6/6/5KB open=2/1/1ms get_next=8/7/7ms]
      -> Redistribution  (cost=0.00..5.07 rows=100000 width=8)
         Hash Key: test1.x
         [40:40 id=100002 dop=40 time=8/7/7ms rows=1289(46/32/20) mem=512/432/0B open=0/0/0ms get_next=8/7/7ms * ]
      -> Local Gather  (cost=0.00..5.01 rows=100000 width=8)
         [id=3 dop=40 time=9/2/0ms rows=1289(1042/32/0) mem=600/600/600B open=0/0/0ms get_next=9/2/0ms local_dop=1/1/1]
         -> Seq Scan on test1  (cost=0.00..5.01 rows=100000 width=8)
            Runtime Filter Target Expr: test1.x
            [id=2 split_count=40 time=11/3/0ms rows=1289(1042/32/0) mem=50512/11849/528B open=11/3/0ms get_next=1/0/0ms scan_rows=100000(8192/7692/1696)]
      -> Hash  (cost=5.00..5.00 rows=1000 width=8)
         [id=8 dop=40 time=2/1/1ms rows=1000(34/25/13) mem=396/396/396KB open=2/1/1ms get_next=0/0/0ms rehash=1/1/1 hash_mem=384/384/384KB]
         -> Redistribution  (cost=0.00..5.00 rows=1000 width=8)
            Hash Key: test2.x
            [40:40 id=100003 dop=40 time=2/1/1ms rows=1000(34/25/13) mem=0/0/0B open=0/0/0ms get_next=2/1/1ms * ]
         -> Local Gather  (cost=0.00..5.00 rows=1000 width=8)
            [id=5 dop=40 time=1/0/0ms rows=1000(1000/25/0) mem=600/600/600B open=1/0/0ms get_next=1/0/0ms local_dop=1/1/1]
            -> Seq Scan on test2  (cost=0.00..5.00 rows=1000 width=8)
               [id=4 split_count=40 time=0/0/0ms rows=1000(1000/25/0) mem=528/528/528B open=0/0/0ms get_next=0/0/0ms scan_rows=1000(1000/1000/1000)]

Los datos del probe side se redistribuyen al operador Hash Join. El motor utiliza automáticamente un filtro en tiempo de ejecución Global para acelerar la consulta.

Ejemplo 4: Filtro In con índice de mapa de bits (V2.2+)

SET hg_experimental_enable_result_cache = OFF;
DROP TABLE IF EXISTS test1, test2;
BEGIN;
CREATE TABLE test1 (x text, y text);
CALL set_table_property('test1', 'distribution_key', 'x');
CALL set_table_property('test1', 'bitmap_columns', 'x');
CALL set_table_property('test1', 'dictionary_encoding_columns', '');
CREATE TABLE test2 (x text, y text);
CALL set_table_property('test2', 'distribution_key', 'x');
END;
INSERT INTO test1 SELECT t::text, t::text FROM generate_series(1, 10000000) t;
INSERT INTO test2 SELECT t::text, t::text FROM generate_series(1, 50) t;
ANALYZE test1;
ANALYZE test2;
EXPLAIN ANALYZE SELECT * FROM test1 JOIN test2 ON test1.x = test2.x;

Plan de ejecución:

QUERY PLAN
Gather  (cost=0.00..11.70 rows=50 width=14)
[40:1 id=100002 dop=1 time=16/16/16ms rows=50(50/50/50) mem=2/2/2KB open=0/0ms get_next=16/16/16ms]
  -> Hash Join  (cost=0.00..11.70 rows=50 width=14)
       Hash Cond: (test1.x = test2.x)
       Runtime Filter Cond: (test1.x = test2.x)
       [id=7 dop=40 time=15/9/3ms rows=50(3/1/0) mem=5132/3774/264B open=1/0/0ms get_next=14/8/3ms]
       -> Local Gather  (cost=0.00..6.26 rows=10000000 width=12)
            [id=3 dop=40 time=14/8/3ms rows=50(3/1/0) mem=600/600/600B open=1/0/0ms get_next=14/8/3ms local_dop=1/1/1]
            -> Seq Scan on test1  (cost=0.00..6.06 rows=10000000 width=12)
                 Runtime Filter Target Expr: test1.x
                 [id=2 split_count=40 time=16/10/5ms rows=50(3/1/0) mem=67544/48945/528B open=16/9/5ms get_next=1/0/0ms scan_rows=7247692(250875/249920/248982) bitmap_used=50]
       -> Hash  (cost=5.00..5.00 rows=50 width=2)
            [id=6 dop=40 time=1/0/0ms rows=61(3/1/1) mem=534/530/521KB open=1/0/0ms get_next=1/0/0ms rehash=1/1/1 hash_mem=512/512/512KB]
            -> Local Gather  (cost=0.00..5.00 rows=50 width=2)
                 [id=5 dop=40 time=1/0/0ms rows=50(3/1/0) mem=600/600/600B open=0/0/0ms get_next=1/0/0ms local_dop=1/1/1]
                 -> Seq Scan on test2  (cost=0.00..5.00 rows=50 width=2)
                      [id=4 split_count=40 time=1/0/0ms rows=50(3/1/0) mem=528/528/528B open=1/0/0ms get_next=0/0/0ms scan_rows=50(3/1/1)]

El operador de escaneo del probe side utiliza un índice de mapa de bits. El filtro In proporciona un filtrado preciso, dejando solo 50 filas. El valor scan_rows en el operador de escaneo supera los 7 millones, cifra inferior a los 10 millones de filas originales. Esto se debe a que el filtro In puede enviarse al motor de almacenamiento, reduciendo la sobrecarga de E/S. Combinar un filtro In con un índice de mapa de bits ofrece ganancias significativas cuando la clave de unión es de tipo STRING.

Ejemplo 5: Filtro MinMax para reducción de E/S (V2.2+)

SET hg_experimental_enable_result_cache = OFF;
DROP TABLE IF EXISTS test1, test2;
BEGIN;
CREATE TABLE test1 (x int, y int);
CALL set_table_property('test1', 'distribution_key', 'x');
CREATE TABLE test2 (x int, y int);
CALL set_table_property('test2', 'distribution_key', 'x');
END;
INSERT INTO test1 SELECT t::int, t::int FROM generate_series(1, 10000000) t;
INSERT INTO test2 SELECT t::int, t::int FROM generate_series(1, 100000) t;
ANALYZE test1;
ANALYZE test2;
EXPLAIN ANALYZE SELECT * FROM test1 JOIN test2 ON test1.x = test2.x;

Plan de ejecución:

QUERY PLAN
 Gather  (cost=0.00..15.68 rows=100000 width=16)
   [40:1 id=100002 dop=1 time=5/5/5ms rows=100000(100000/100000/100000) mem=600/600/600B open=0/0/0ms get_next=5/5/5ms]
   -> Hash Join  (cost=0.00..11.98 rows=100000 width=16)
        Hash Cond: (test1.x = test2.x)
        Runtime Filter Cond: (test1.x = test2.x)
        [id=7 dop=40 time=5/4/4ms rows=100000(2639/2500/2406) mem=97/92/89KB open=1/0/0ms get_next=4/3/3ms]
        -> Local Gather  (cost=0.00..6.14 rows=10000000 width=8)
             [id=3 dop=40 time=5/3/3ms rows=100000(2639/2500/2406) mem=600/600/600B open=1/0/0ms get_next=4/3/3ms local_dop=1/1/1]
             -> Seq Scan on test1  (cost=0.00..6.00 rows=10000000 width=8)
                  Runtime Filter Target Expr: test1.x
                  [id=2 split_count=40 time=6/6/5ms rows=100000(2639/2500/2406) mem=61/60/59KB open=6/5/5ms get_next=0/0/0ms scan_rows=327680(8192/8192/8192)]
        -> Hash  (cost=5.01..5.01 rows=100000 width=8)
             [id=6 dop=40 time=1/0/0ms rows=100000(2639/2500/2406) mem=463/460/458KB open=1/0/0ms get_next=1/0/0ms rehash=1/1/1 hash_mem=384/384/384KB]
             -> Local Gather  (cost=0.00..5.01 rows=100000 width=8)
                  [id=5 dop=40 time=1/0/0ms rows=100000(2639/2500/2406) mem=600/600/600B open=0/0/0ms get_next=1/0/0ms local_dop=1/1/1]
                  -> Seq Scan on test2  (cost=0.00..5.01 rows=100000 width=8)
                       [id=4 split_count=40 time=1/0/0ms rows=100000(2639/2500/2406) mem=528/528/528B open=1/0/0ms get_next=0/0/0ms scan_rows=100000(2639/2500/2406)]

El operador de escaneo del probe side lee poco más de 320,000 filas del motor de almacenamiento, una cifra muy inferior a los 10 millones originales. Esto ocurre porque el filtro en tiempo de ejecución se envía al motor de almacenamiento, que utiliza los metadatos de un lote de datos para filtrar lotes completos a la vez. Esto puede reducir significativamente la sobrecarga de E/S. Este tipo de filtro es más efectivo cuando la clave de unión es numérica y el rango de valores del build side es menor que el rango del probe side.

Ejemplo 6: Filtro en tiempo de ejecución TopN (V4.0+)

Cuando una sentencia SQL incluye un operador topN, Hologres no calcula todos los resultados. En su lugar, genera un filtro dinámico para descartar datos de forma temprana.

SELECT o_orderkey FROM orders ORDER BY o_orderdate LIMIT 5;

Plan de ejecución:

QUERY PLAN
Limit  (cost=0.00..116554.70 rows=0 width=8)
  ->  Sort  (cost=0.00..116554.70 rows=100 width=12)
        Sort Key: o_orderdate
      [id=6 dop=1 time=317/317/317ms rows=5(5/5/5) mem=1/1/1KB open=317/317/317ms get_next=0/0/0ms]
        ->  Gather  (cost=0.00..116554.25 rows=100 width=12)
            [20:1 id=100002 dop=1 time=317/317/317ms rows=100(100/100/100) mem=6/6/6KB open=0/0/0ms get_next=317/317/317ms * ]
              ->  Limit  (cost=0.00..116554.25 rows=0 width=12)
                    ->  Sort  (cost=0.00..116554.25 rows=150000000 width=12)
                          Sort Key: o_orderdate
                          Runtime Filter Sort Column: o_orderdate
                        [id=3 dop=20 time=318/282/258ms rows=100(5/5/5) mem=96/96/96KB open=318/282/258ms get_next=1/0/0ms]
                          ->  Local Gather  (cost=0.00..9.59 rows=150000000 width=12)
                              [id=2 dop=20 time=316/280/256ms rows=1372205(68691/68610/68498) mem=0/0/0B open=0/0/0ms get_next=316/280/256ms local_dop=1/1/1 * ]
                                ->  Seq Scan on orders  (cost=0.00..8.24 rows=150000000 width=12)
                                      Runtime Filter Target Expr: o_orderdate
                                    [id=1 split_count=20 time=286/249/222ms rows=1372205(68691/68610/68498) mem=179/179/179KB open=0/0/0ms get_next=286/249/222ms physical_reads=27074(1426/1353/1294) scan_rows=144867963(7324934/7243398/7172304)]
Query id:[1001003033996040311]
QE version: 2.0
Query Queue: init_warehouse.default_queue
======================cost======================
Total cost:[343] ms
Optimizer cost:[13] ms
Build execution plan cost:[0] ms
Init execution plan cost:[6] ms
Start query cost:[6] ms
- Queue cost: [0] ms
- Wait schema cost:[0] ms
- Lock query cost:[0] ms
- Create dataset reader cost:[0] ms
- Create split reader cost:[0] ms
Get result cost:[318] ms
- Get the first block cost:[318] ms
====================resource====================
Memory: total 7 MB. Worker stats: max 3 MB, avg 3 MB, min 3 MB, max memory worker id: 189*****.
CPU time: total 5167 ms. Worker stats: max 2610 ms, avg 2583 ms, min 2557 ms, max CPU time worker id: 189*****.
DAG CPU time stats: max 5165 ms, avg 2582 ms, min 0 ms, cnt 2, max CPU time dag id: 1.
Fragment CPU time stats: max 5137 ms, avg 1721 ms, min 0 ms, cnt 3, max CPU time fragment id: 2.
Ec wait time: total 90 ms. Worker stats: max 46 ms, max(max) 2 ms, avg 45 ms, min 44 ms, max ec wait time worker id: 189*****, max(max) ec wait time worker id: 189*****.
Physical read bytes: total 799 MB. Worker stats: max 400 MB, avg 399 MB, min 399 MB, max physical read bytes worker id: 189*****.
Read bytes: total 898 MB. Worker stats: max 450 MB, avg 449 MB, min 448 MB, max read bytes worker id: 189*****.
DAG instance count: total 3. Worker stats: max 2, avg 1, min 1, max DAG instance count worker id: 189*****.
Fragment instance count: total 41. Worker stats: max 21, avg 20, min 20, max fragment instance count worker id: 189*****.

Sin el filtro en tiempo de ejecución TopN, el ScanNode leería cada bloque de datos de la tabla de pedidos y lo pasaría al nodo TopN, que luego usaría una ordenación por montículos para mantener las 5 filas principales vistas hasta el momento.

Por ejemplo, cada bloque de datos contiene aproximadamente 8,192 filas. Después de procesar el primer bloque, TopN conoce la o_orderdate clasificada en quinto lugar en ese bloque. Supongamos que es 1995-01-01. Cuando el nodo Scan lee el segundo bloque, utiliza 1995-01-01 como condición de filtro y envía solo las filas donde o_orderdate <= 1995-01-01 a TopN. El umbral se actualiza dinámicamente. Si la o_orderdate clasificada en quinto lugar en el segundo bloque es menor, TopN reemplaza el umbral anterior con el nuevo valor.

Puede usar la salida de EXPLAIN para ver el filtro en tiempo de ejecución TopN generado por el optimizador:

->  Limit  (cost=0.00..116554.25 rows=0 width=12)
  ->  Sort  (cost=0.00..116554.25 rows=150000000 width=12)
        Sort Key: o_orderdate
        Runtime Filter Sort Column: o_orderdate
      [id=3 dop=20 time=318/282/258ms rows=100(5/5/5) mem=96/96/96KB open=318/282/258ms get_next=1/0/0ms]

La presencia de Runtime Filter Sort Column en el nodo TopN indica que este nodo genera un filtro en tiempo de ejecución TopN.

Ejemplo 7: Cross join con ScalarFilter (V4.2+)

SET hg_experimental_enable_result_cache = OFF;
DROP TABLE IF EXISTS t1, t2;
BEGIN;
CREATE TABLE t1 (a int, b int);
CREATE TABLE t2 (a int, b int);
END;
INSERT INTO t1 SELECT t, t FROM generate_series(1, 1000000) t;
INSERT INTO t2 SELECT t, t FROM generate_series(1, 1000) t;
ANALYZE t1;
ANALYZE t2;
EXPLAIN ANALYZE
SELECT * FROM t1
WHERE a >= (SELECT min(a) FROM t2 WHERE b BETWEEN 0 AND 1)
  AND a <= (SELECT max(a) FROM t2 WHERE b BETWEEN 0 AND 1);

Puntos destacados del plan de ejecución:

El nodo Cross Join muestra Runtime Filter Build Expr: (min(t2.a)), (max(t2.a)).
El ScanNode en t1 muestra Runtime Filter Target Expr: (t1.a >= ${1}) AND (t1.a <= ${2}).
Para t1, scan_rows refleja el número de filas leídas del almacenamiento, mientras que rows disminuye significativamente después de aplicar el ScalarFilter. Esta diferencia verifica el efecto del filtro.

Deshabilitar la función para comparar:

SET hg_experimental_generate_runtime_scalar_filter = off;
EXPLAIN ANALYZE
SELECT * FROM t1
WHERE a >= (SELECT min(a) FROM t2 WHERE b BETWEEN 0 AND 1)
  AND a <= (SELECT max(a) FROM t2 WHERE b BETWEEN 0 AND 1);

Después de deshabilitar la función, el plan de ejecución ya no muestra los campos Runtime Filter Build Expr y Target Expr, y el escaneo en t1 vuelve a ser un escaneo completo de la tabla.

Historial de versiones

Versión	Nuevas funciones
V2.0	Compatibilidad con filtros en tiempo de ejecución en Hash Join (tipo Local, incluidos filtros Bloom, In y MinMax).
V2.1	Compatibilidad con filtros en tiempo de ejecución con condiciones de unión de múltiples columnas.
V2.2	Compatibilidad con filtros en tiempo de ejecución Global (para uniones con redistribución); los filtros In pueden combinarse con índices de mapa de bits.
V4.0	Compatibilidad con filtro en tiempo de ejecución TopN.
V4.2	Compatibilidad con filtros en tiempo de ejecución Cross Join (ScalarFilter) para optimizar escenarios que utilizan una subconsulta escalar para filtrar una tabla grande.