code: def tumble_window_demo():
      env = StreamExecutionEnvironment.get_execution_environment()
      env.set_parallelism(1)
      t_env = StreamTableEnvironment.create(
          stream_execution_environment=env
      )
  
      # define the source with watermark definition
      ds = env.from_collection(
          collection=[
              (Instant.of_epoch_milli(1000), "Alice", 110.1),
              (Instant.of_epoch_milli(4000), "Bob", 30.2),
              (Instant.of_epoch_milli(3000), "Alice", 20.0),
              (Instant.of_epoch_milli(2000), "Bob", 53.1),
              (Instant.of_epoch_milli(5000), "Alice", 13.1),
              (Instant.of_epoch_milli(3000), "Bob", 3.1),
              (Instant.of_epoch_milli(7000), "Bob", 16.1),
              (Instant.of_epoch_milli(10000), "Alice", 20.1),
          ],
          type_info=Types.ROW(
              [Types.INSTANT(), Types.STRING(), Types.FLOAT()]
          ),
      )
  
      table = t_env.from_data_stream(
          ds,
          Schema.new_builder()
          .column_by_expression("ts", "CAST(f0 AS TIMESTAMP(3))")
          .column("f1", DataTypes.STRING())
          .column("f2", DataTypes.FLOAT())
          .watermark("ts", "ts - INTERVAL '3' SECOND")
          .build(),
      ).alias("ts", "name", "price")
  
      # define the sink
      t_env.create_temporary_table(
          "sink",
          TableDescriptor.for_connector("print")
          .schema(
              Schema.new_builder()
              .column("name", DataTypes.STRING())
              .column("total_price", DataTypes.FLOAT())
              .build()
          )
          .build(),
      )
  
      # define the over window operation
      table = table.over_window(
          Over.partition_by(col("name"))
          .order_by(col("ts"))
          .preceding(row_interval(2))
          .following(CURRENT_ROW)
          .alias("w")
      ).select(col("name"), col("price").max.over(col("w")))
  
      # submit for execution
      table.execute_insert("sink").wait()
  
file_path: flink/flink-python/pyflink/examples/table/windowing/over_window.py
imports: import logging
  import sys
  from pyflink.common.time import Instant
  from pyflink.common import Types
  from pyflink.datastream import StreamExecutionEnvironment
  from pyflink.table import DataTypes, TableDescriptor, Schema, StreamTableEnvironment
  from pyflink.table.expressions import col, row_interval, CURRENT_ROW
  from pyflink.table.window import Over
label_desc: Use image node_id 1 for calling a global function with example usage: tumble_window_demo() without return types