table_env.execute_sql("""
SELECT 
user_id,
pickupgeohash,
dropgeohash,
city,
category,
dt,
hr,
mnt,
window_time as rowtime,
`5_mnt_btch`,
ride_distance
FROM (
   SELECT *,
     ROW_NUMBER() OVER (PARTITION BY user_id,pickupgeohash,window_start, window_end,window_time ORDER BY rowtime desc) AS rownum
   FROM  TABLE(TUMBLE(TABLE sql_full_df, DESCRIPTOR(rowtime), INTERVAL '5' MINUTES)))  
WHERE rownum = 1
""" ).print()