Hi, I'm new to Flink and the channel so not sure if I'm at the right place. I have been encountering some performance issues with a specific query. I'm working with two tables - one table is called

bl

, which is relatively small with about 100 rows (wanted to add more to the stream later)

CREATE TABLE IF NOT EXISTS bl (
    inserted_on_epoch BIGINT,
    id INT,
    timestamp_epoch BIGINT,
    fixture_id INT,
    market_type_id INT,
    segment_id INT,
    participant_id INT,
    side_id INT,
    caption STRING,
    metadata_ts TIMESTAMP(3) METADATA FROM 'timestamp',
    metadata_offset INT METADATA FROM 'offset'
 ) WITH (

   'connector' = 'kafka',
   ...
)

And a second table called

m

, which is much larger, approximately 10 GB in size.

CREATE TABLE m (
    market_id INT,
    source_id INT,
    fixture_id INT,
    market_type_id INT,
    segment_id INT,
    price DECIMAL(10, 2),
    inserted_on_epoch BIGINT,
    uuid STRING,
    side_id INT
) WITH (
    'connector' = 'jdbc',
    ...
)

And I have this query to join the two tables:

SELECT *,
       ROW_NUMBER() OVER (PARTITION BY bl.id ORDER BY m.inserted_on_epoch DESC) AS row_num
FROM bl
INNER JOIN m USING (fixture_id, market_type_id, segment_id, side_id)
WHERE m.inserted_on_epoch < bl.timestamp_epoch
)
WHERE row_num = 1

When I run the above query in my Postgres DB, it finishes in just a few seconds. However, when executing the same operation in Flink, it takes hours for the query to complete (to have the same result as the result from running the query on the postgres DB). I have already tried increasing parallelism, which has resulted in some improvement, but the query execution time is still much longer than what I get with postgres DB. Is there anything else I could try to improve the performance of this query in Flink? I'm not sure what I could be overlooking and would appreciate any advice or suggestions!