¿Qué tan eficiente es el estado utilizado para las consultas de top-n en Apache Flink?

Me pregunto cuánto estado utiliza Apache Flink para consultas Top-N y tablas en general.

Primero, estoy utilizando Flink SQL para procesar mensajes de un tema de Kafka:

CREATE TABLE purchases (
  country STRING,
  product STRING
) WITH (
   'connector' = 'kafka',
   'topic' = 'purchases',
   'properties.bootstrap.servers' = 'kafka:29092',
   'value.format' = 'json',
   'properties.group.id' = '1',
   'scan.startup.mode' = 'earliest-offset'
);

También inicialicé un conector JDBC:

CREATE TABLE aggregations (
  `country` STRING,
  `product` STRING,
  `purchases` BIGINT NOT NULL,
  PRIMARY KEY (`country`, `product`) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:postgresql://postgres:5432/postgres?&user=postgres&password=postgres',
  'table-name' = 'aggregations'
);

Finalmente, inicio la agregación:

insert into aggregations
SELECT `country`, `product`, `purchases`
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY country ORDER BY `purchases` DESC) AS row_num
  FROM (select country, product, count(*) as `purchases` from purchases group by country, product))
WHERE row_num <= 3;

Desde la documentación de administración de estado de Flink, dice:

Conceptualmente, las tablas fuente nunca se mantienen enteramente en un estado. Un desarrollador maneja tablas lógicas (es decir, tablas dinámicas). Sus requisitos de estado dependen de las operaciones utilizadas.

Entonces, ¿entiendo correctamente que Flink no guarda las filas de la tabla “purchases” del conector de Kafka?

Más importante aún, en la agregación:

select country, product, count(*) as `purchases` from purchases group by country, product

¿Flink mantiene cada clave de país y producto en el estado?