title: How much state is used for top-n queries in Apache Flink?
body: <p>I'm wondering how much state <a href="https://flink.apache.org" rel="nofollow noreferrer">Apache Flink</a> uses for <a href="https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/queries/topn/" rel="nofollow noreferrer">Top-N</a> queries and tables in general.</p>
  <p>First, I'm using Flink SQL to process messages from a Kafka topic:</p>
  <pre class="lang-sql prettyprint-override"><code>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'
  );
  </code></pre>
  <p>I also initialized a JDBC connector:</p>
  <pre class="lang-sql prettyprint-override"><code>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?&amp;user=postgres&amp;password=postgres',
    'table-name' = 'aggregations'
  );
  </code></pre>
  <p>Finally I start the aggregation:</p>
  <pre class="lang-sql prettyprint-override"><code>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 &lt;= 3;
  </code></pre>
  <p>From Flink state management <a href="https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/concepts/overview/" rel="nofollow noreferrer">docs</a> is says:</p>
  <blockquote>
  <p>Conceptually, source tables are never kept entirely in state. An implementer deals with logical tables (i.e. dynamic tables). Their state requirements depend on the used operations.</p>
  </blockquote>
  <p>So do I understand correctly that Flink doesn't save the rows of <code>purchases</code> table from Kafka connector?</p>
  <p>More importantly, in aggregation:</p>
  <pre class="lang-sql prettyprint-override"><code>select country, product, count(*) as `purchases` from purchases group by country, product
  </code></pre>
  <p>does Flink keep every country, product key in state?</p>