flaky test: Failed to build executor due to Expired Epoch

[xxchan]

Describe the bug

https://buildkite.com/risingwavelabs/pull-request/builds/9627#0183a1f8-4d62-4d37-b22a-8aaeed6996a5

Error info

failed to run `e2e_test/streaming/demo/ad_ctr.slt` --   |     | Caused by:   | query failed: db error: ERROR: QueryError: internal error: [PlanNode: Ok(Exchange(ExchangeNode { sources: [ExchangeSource { task_output_id: Some(TaskOutputId { task_id: Some(TaskId { query_id: "587e7934-8a8a-487c-8645-f78d39a0094f", stage_id: 1, task_id: 0 }), output_id: 0 }), host: Some(HostAddress { host: "127.0.0.1", port: 5688 }), local_execute_plan: Some(Plan(LocalExecutePlan { plan: Some(PlanFragment { root: Some(PlanNode { children: [PlanNode { children: [], identity: "3195bdd1-1dea-4f54-a588-d6a444d69e10", node_body: Some(RowSeqScan(RowSeqScanNode { table_desc: Some(StorageTableDesc { table_id: 1221, columns: [ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1000, name: "ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1001, name: "ctr", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1002, name: "window_end", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1003, name: "ad_impression.ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1004, name: "(TumbleStart(ad_impression.impression_timestamp, '00:01:00':Interval) + '00:01:00':Interval)", field_descs: [], type_name: "" }], pk: [ColumnOrder { order_type: Ascending, index: 3 }, ColumnOrder { order_type: Ascending, index: 2 }, ColumnOrder { order_type: Ascending, index: 0 }, ColumnOrder { order_type: Ascending, index: 4 }], dist_key_indices: [3, 2], retention_seconds: 0, value_indices: [0, 1, 2, 3, 4] }), column_ids: [1000, 1001, 1002], scan_ranges: [], vnode_bitmap: Some(Buffer { compression: None, body: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 224, 255, 255, 3, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] }) })) }], identity: "02e4eecd-bd9c-4170-89e5-0af10a558ebe", node_body: Some(Project(ProjectNode { select_list: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 0 })) }, ExprNode { expr_type: RoundDigit, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(FuncCall(FunctionCall { children: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 1 })) }, ExprNode { expr_type: ConstantValue, return_type: Some(DataType { type_name: Int32, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(Constant(ConstantValue { body: [0, 0, 0, 2] })) }] })) }, ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 2 })) }] })) }), exchange_info: Some(ExchangeInfo { mode: Single, distribution: None }) }), epoch: 3121804801998848 })) }, ExchangeSource { task_output_id: Some(TaskOutputId { task_id: Some(TaskId { query_id: "587e7934-8a8a-487c-8645-f78d39a0094f", stage_id: 1, task_id: 1 }), output_id: 0 }), host: Some(HostAddress { host: "127.0.0.1", port: 5687 }), local_execute_plan: Some(Plan(LocalExecutePlan { plan: Some(PlanFragment { root: Some(PlanNode { children: [PlanNode { children: [], identity: "b6e7e288-911a-4a4b-a9e9-8932f6c2fa86", node_body: Some(RowSeqScan(RowSeqScanNode { table_desc: Some(StorageTableDesc { table_id: 1221, columns: [ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1000, name: "ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1001, name: "ctr", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1002, name: "window_end", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1003, name: "ad_impression.ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1004, name: "(TumbleStart(ad_impression.impression_timestamp, '00:01:00':Interval) + '00:01:00':Interval)", field_descs: [], type_name: "" }], pk: [ColumnOrder { order_type: Ascending, index: 3 }, ColumnOrder { order_type: Ascending, index: 2 }, ColumnOrder { order_type: Ascending, index: 0 }, ColumnOrder { order_type: Ascending, index: 4 }], dist_key_indices: [3, 2], retention_seconds: 0, value_indices: [0, 1, 2, 3, 4] }), column_ids: [1000, 1001, 1002], scan_ranges: [], vnode_bitmap: Some(Buffer { compression: None, body: [0, 0, 0, 0, 0, 0, 0, 0, 0, 252, 255, 255, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] }) })) }], identity: "31756088-4047-479c-aa8e-3e5e51a43ad9", node_body: Some(Project(ProjectNode { select_list: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 0 })) }, ExprNode { expr_type: RoundDigit, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(FuncCall(FunctionCall { children: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 1 })) }, ExprNode { expr_type: ConstantValue, return_type: Some(DataType { type_name: Int32, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(Constant(ConstantValue { body: [0, 0, 0, 2] })) }] })) }, ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 2 })) }] })) }), exchange_info: Some(ExchangeInfo { mode: Single, distribution: None }) }), epoch: 3121804801998848 })) }, ExchangeSource { task_output_id: Some(TaskOutputId { task_id: Some(TaskId { query_id: "587e7934-8a8a-487c-8645-f78d39a0094f", stage_id: 1, task_id: 2 }), output_id: 0 }), host: Some(HostAddress { host: "127.0.0.1", port: 5689 }), local_execute_plan: Some(Plan(LocalExecutePlan { plan: Some(PlanFragment { root: Some(PlanNode { children: [PlanNode { children: [], identity: "f04a3496-65f5-4a54-9b23-12eb52c0f8f3", node_body: Some(RowSeqScan(RowSeqScanNode { table_desc: Some(StorageTableDesc { table_id: 1221, columns: [ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1000, name: "ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1001, name: "ctr", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1002, name: "window_end", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1003, name: "ad_impression.ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1004, name: "(TumbleStart(ad_impression.impression_timestamp, '00:01:00':Interval) + '00:01:00':Interval)", field_descs: [], type_name: "" }], pk: [ColumnOrder { order_type: Ascending, index: 3 }, ColumnOrder { order_type: Ascending, index: 2 }, ColumnOrder { order_type: Ascending, index: 0 }, ColumnOrder { order_type: Ascending, index: 4 }], dist_key_indices: [3, 2], retention_seconds: 0, value_indices: [0, 1, 2, 3, 4] }), column_ids: [1000, 1001, 1002], scan_ranges: [], vnode_bitmap: Some(Buffer { compression: None, body: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 248, 255, 255] }) })) }], identity: "5c40d7ab-d9bc-40e2-83d0-abaeea532510", node_body: Some(Project(ProjectNode { select_list: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 0 })) }, ExprNode { expr_type: RoundDigit, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(FuncCall(FunctionCall { children: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 1 })) }, ExprNode { expr_type: ConstantValue, return_type: Some(DataType { type_name: Int32, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(Constant(ConstantValue { body: [0, 0, 0, 2] })) }] })) }, ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 2 })) }] })) }), exchange_info: Some(ExchangeInfo { mode: Single, distribution: None }) }), epoch: 3121804801998848 })) }, ExchangeSource { task_output_id: Some(TaskOutputId { task_id: Some(TaskId { query_id: "587e7934-8a8a-487c-8645-f78d39a0094f", stage_id: 1, task_id: 3 }), output_id: 0 }), host: Some(HostAddress { host: "127.0.0.1", port: 5688 }), local_execute_plan: Some(Plan(LocalExecutePlan { plan: Some(PlanFragment { root: Some(PlanNode { children: [PlanNode { children: [], identity: "9a4dfbfc-242b-4dc2-8c9e-525692bf2432", node_body: Some(RowSeqScan(RowSeqScanNode { table_desc: Some(StorageTableDesc { table_id: 1221, columns: [ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1000, name: "ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1001, name: "ctr", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1002, name: "window_end", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1003, name: "ad_impression.ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1004, name: "(TumbleStart(ad_impression.impression_timestamp, '00:01:00':Interval) + '00:01:00':Interval)", field_descs: [], type_name: "" }], pk: [ColumnOrder { order_type: Ascending, index: 3 }, ColumnOrder { order_type: Ascending, index: 2 }, ColumnOrder { order_type: Ascending, index: 0 }, ColumnOrder { order_type: Ascending, index: 4 }], dist_key_indices: [3, 2], retention_seconds: 0, value_indices: [0, 1, 2, 3, 4] }), column_ids: [1000, 1001, 1002], scan_ranges: [], vnode_bitmap: Some(Buffer { compression: None, body: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 255, 255, 31, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] }) })) }], identity: "e71b22ac-65e0-42d2-80de-5c9fd037ec04", node_body: Some(Project(ProjectNode { select_list: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 0 })) }, ExprNode { expr_type: RoundDigit, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(FuncCall(FunctionCall { children: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 1 })) }, ExprNode { expr_type: ConstantValue, return_type: Some(DataType { type_name: Int32, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(Constant(ConstantValue { body: [0, 0, 0, 2] })) }] })) }, ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 2 })) }] })) }), exchange_info: Some(ExchangeInfo { mode: Single, distribution: None }) }), epoch: 3121804801998848 })) }, ExchangeSource { task_output_id: Some(TaskOutputId { task_id: Some(TaskId { query_id: "587e7934-8a8a-487c-8645-f78d39a0094f", stage_id: 1, task_id: 4 }), output_id: 0 }), host: Some(HostAddress { host: "127.0.0.1", port: 5687 }), local_execute_plan: Some(Plan(LocalExecutePlan { plan: Some(PlanFragment { root: Some(PlanNode { children: [PlanNode { children: [], identity: "5639f19c-20e1-4eee-8ba0-60b73e522cd8", node_body: Some(RowSeqScan(RowSeqScanNode { table_desc: Some(StorageTableDesc { table_id: 1221, columns: [ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1000, name: "ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1001, name: "ctr", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1002, name: "window_end", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1003, name: "ad_impression.ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1004, name: "(TumbleStart(ad_impression.impression_timestamp, '00:01:00':Interval) + '00:01:00':Interval)", field_descs: [], type_name: "" }], pk: [ColumnOrder { order_type: Ascending, index: 3 }, ColumnOrder { order_type: Ascending, index: 2 }, ColumnOrder { order_type: Ascending, index: 0 }, ColumnOrder { order_type: Ascending, index: 4 }], dist_key_indices: [3, 2], retention_seconds: 0, value_indices: [0, 1, 2, 3, 4] }), column_ids: [1000, 1001, 1002], scan_ranges: [], vnode_bitmap: Some(Buffer { compression: None, body: [0, 0, 0, 0, 0, 0, 240, 255, 255, 3, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] }) })) }], identity: "fadd60e9-f969-44b0-b177-58ba6d6d0904", node_body: Some(Project(ProjectNode { select_list: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 0 })) }, ExprNode { expr_type: RoundDigit, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(FuncCall(FunctionCall { children: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 1 })) }, ExprNode { expr_type: ConstantValue, return_type: Some(DataType { type_name: Int32, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(Constant(ConstantValue { body: [0, 0, 0, 2] })) }] })) }, ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 2 })) }] })) }), exchange_info: Some(ExchangeInfo { mode: Single, distribution: None }) }), epoch: 3121804801998848 })) }, ExchangeSource { task_output_id: Some(TaskOutputId { task_id: Some(TaskId { query_id: "587e7934-8a8a-487c-8645-f78d39a0094f", stage_id: 1, task_id: 5 }), output_id: 0 }), host: Some(HostAddress { host: "127.0.0.1", port: 5689 }), local_execute_plan: Some(Plan(LocalExecutePlan { plan: Some(PlanFragment { root: Some(PlanNode { children: [PlanNode { children: [], identity: "e9a99688-4ded-486b-81e8-4c27fdbfe9b6", node_body: Some(RowSeqScan(RowSeqScanNode { table_desc: Some(StorageTableDesc { table_id: 1221, columns: [ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1000, name: "ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1001, name: "ctr", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1002, name: "window_end", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1003, name: "ad_impression.ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1004, name: "(TumbleStart(ad_impression.impression_timestamp, '00:01:00':Interval) + '00:01:00':Interval)", field_descs: [], type_name: "" }], pk: [ColumnOrder { order_type: Ascending, index: 3 }, ColumnOrder { order_type: Ascending, index: 2 }, ColumnOrder { order_type: Ascending, index: 0 }, ColumnOrder { order_type: Ascending, index: 4 }], dist_key_indices: [3, 2], retention_seconds: 0, value_indices: [0, 1, 2, 3, 4] }), column_ids: [1000, 1001, 1002], scan_ranges: [], vnode_bitmap: Some(Buffer { compression: None, body: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 240, 255, 255, 1, 0, 0, 0, 0, 0, 0, 0] }) })) }], identity: "950f427c-84ca-4204-a06f-4bba3cefa9a0", node_body: Some(Project(ProjectNode { select_list: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 0 })) }, ExprNode { expr_type: RoundDigit, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(FuncCall(FunctionCall { children: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 1 })) }, ExprNode { expr_type: ConstantValue, return_type: Some(DataType { type_name: Int32, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(Constant(ConstantValue { body: [0, 0, 0, 2] })) }] })) }, ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 2 })) }] })) }), exchange_info: Some(ExchangeInfo { mode: Single, distribution: None }) }), epoch: 3121804801998848 })) }, ExchangeSource { task_output_id: Some(TaskOutputId { task_id: Some(TaskId { query_id: "587e7934-8a8a-487c-8645-f78d39a0094f", stage_id: 1, task_id: 6 }), output_id: 0 }), host: Some(HostAddress { host: "127.0.0.1", port: 5689 }), local_execute_plan: Some(Plan(LocalExecutePlan { plan: Some(PlanFragment { root: Some(PlanNode { children: [PlanNode { children: [], identity: "9ded7d19-1364-433a-862e-7e71ac5d4bcc", node_body: Some(RowSeqScan(RowSeqScanNode { table_desc: Some(StorageTableDesc { table_id: 1221, columns: [ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1000, name: "ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1001, name: "ctr", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1002, name: "window_end", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1003, name: "ad_impression.ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1004, name: "(TumbleStart(ad_impression.impression_timestamp, '00:01:00':Interval) + '00:01:00':Interval)", field_descs: [], type_name: "" }], pk: [ColumnOrder { order_type: Ascending, index: 3 }, ColumnOrder { order_type: Ascending, index: 2 }, ColumnOrder { order_type: Ascending, index: 0 }, ColumnOrder { order_type: Ascending, index: 4 }], dist_key_indices: [3, 2], retention_seconds: 0, value_indices: [0, 1, 2, 3, 4] }), column_ids: [1000, 1001, 1002], scan_ranges: [], vnode_bitmap: Some(Buffer { compression: None, body: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 192, 255, 255, 7, 0, 0] }) })) }], identity: "7afb094b-e525-42d9-84c9-ac12dcbec28b", node_body: Some(Project(ProjectNode { select_list: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 0 })) }, ExprNode { expr_type: RoundDigit, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(FuncCall(FunctionCall { children: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 1 })) }, ExprNode { expr_type: ConstantValue, return_type: Some(DataType { type_name: Int32, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(Constant(ConstantValue { body: [0, 0, 0, 2] })) }] })) }, ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 2 })) }] })) }), exchange_info: Some(ExchangeInfo { mode: Single, distribution: None }) }), epoch: 3121804801998848 })) }, ExchangeSource { task_output_id: Some(TaskOutputId { task_id: Some(TaskId { query_id: "587e7934-8a8a-487c-8645-f78d39a0094f", stage_id: 1, task_id: 7 }), output_id: 0 }), host: Some(HostAddress { host: "127.0.0.1", port: 5687 }), local_execute_plan: Some(Plan(LocalExecutePlan { plan: Some(PlanFragment { root: Some(PlanNode { children: [PlanNode { children: [], identity: "a1fe0d23-be3c-41e2-a01f-bc51e372195c", node_body: Some(RowSeqScan(RowSeqScanNode { table_desc: Some(StorageTableDesc { table_id: 1221, columns: [ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1000, name: "ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1001, name: "ctr", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1002, name: "window_end", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1003, name: "ad_impression.ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1004, name: "(TumbleStart(ad_impression.impression_timestamp, '00:01:00':Interval) + '00:01:00':Interval)", field_descs: [], type_name: "" }], pk: [ColumnOrder { order_type: Ascending, index: 3 }, ColumnOrder { order_type: Ascending, index: 2 }, ColumnOrder { order_type: Ascending, index: 0 }, ColumnOrder { order_type: Ascending, index: 4 }], dist_key_indices: [3, 2], retention_seconds: 0, value_indices: [0, 1, 2, 3, 4] }), column_ids: [1000, 1001, 1002], scan_ranges: [], vnode_bitmap: Some(Buffer { compression: None, body: [0, 255, 255, 63, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] }) })) }], identity: "66b2a1bd-63df-43c2-a525-93b83fc3c3b9", node_body: Some(Project(ProjectNode { select_list: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 0 })) }, ExprNode { expr_type: RoundDigit, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(FuncCall(FunctionCall { children: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 1 })) }, ExprNode { expr_type: ConstantValue, return_type: Some(DataType { type_name: Int32, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(Constant(ConstantValue { body: [0, 0, 0, 2] })) }] })) }, ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 2 })) }] })) }), exchange_info: Some(ExchangeInfo { mode: Single, distribution: None }) }), epoch: 3121804801998848 })) }, ExchangeSource { task_output_id: Some(TaskOutputId { task_id: Some(TaskId { query_id: "587e7934-8a8a-487c-8645-f78d39a0094f", stage_id: 1, task_id: 8 }), output_id: 0 }), host: Some(HostAddress { host: "127.0.0.1", port: 5688 }), local_execute_plan: Some(Plan(LocalExecutePlan { plan: Some(PlanFragment { root: Some(PlanNode { children: [PlanNode { children: [], identity: "e2fbbaa2-b037-448a-933f-514b0f5c36db", node_body: Some(RowSeqScan(RowSeqScanNode { table_desc: Some(StorageTableDesc { table_id: 1221, columns: [ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1000, name: "ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1001, name: "ctr", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1002, name: "window_end", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1003, name: "ad_impression.ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1004, name: "(TumbleStart(ad_impression.impression_timestamp, '00:01:00':Interval) + '00:01:00':Interval)", field_descs: [], type_name: "" }], pk: [ColumnOrder { order_type: Ascending, index: 3 }, ColumnOrder { order_type: Ascending, index: 2 }, ColumnOrder { order_type: Ascending, index: 0 }, ColumnOrder { order_type: Ascending, index: 4 }], dist_key_indices: [3, 2], retention_seconds: 0, value_indices: [0, 1, 2, 3, 4] }), column_ids: [1000, 1001, 1002], scan_ranges: [], vnode_bitmap: Some(Buffer { compression: None, body: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 252, 255, 127, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] }) })) }], identity: "0d140343-d17f-4d71-be68-1a4c3145836b", node_body: Some(Project(ProjectNode { select_list: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 0 })) }, ExprNode { expr_type: RoundDigit, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(FuncCall(FunctionCall { children: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 1 })) }, ExprNode { expr_type: ConstantValue, return_type: Some(DataType { type_name: Int32, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(Constant(ConstantValue { body: [0, 0, 0, 2] })) }] })) }, ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 2 })) }] })) }), exchange_info: Some(ExchangeInfo { mode: Single, distribution: None }) }), epoch: 3121804801998848 })) }, ExchangeSource { task_output_id: Some(TaskOutputId { task_id: Some(TaskId { query_id: "587e7934-8a8a-487c-8645-f78d39a0094f", stage_id: 1, task_id: 9 }), output_id: 0 }), host: Some(HostAddress { host: "127.0.0.1", port: 5688 }), local_execute_plan: Some(Plan(LocalExecutePlan { plan: Some(PlanFragment { root: Some(PlanNode { children: [PlanNode { children: [], identity: "863f4e70-a221-48ba-bbb8-23ea6370f24a", node_body: Some(RowSeqScan(RowSeqScanNode { table_desc: Some(StorageTableDesc { table_id: 1221, columns: [ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1000, name: "ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1001, name: "ctr", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1002, name: "window_end", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1003, name: "ad_impression.ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1004, name: "(TumbleStart(ad_impression.impression_timestamp, '00:01:00':Interval) + '00:01:00':Interval)", field_descs: [], type_name: "" }], pk: [ColumnOrder { order_type: Ascending, index: 3 }, ColumnOrder { order_type: Ascending, index: 2 }, ColumnOrder { order_type: Ascending, index: 0 }, ColumnOrder { order_type: Ascending, index: 4 }], dist_key_indices: [3, 2], retention_seconds: 0, value_indices: [0, 1, 2, 3, 4] }), column_ids: [1000, 1001, 1002], scan_ranges: [], vnode_bitmap: Some(Buffer { compression: None, body: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 128, 255, 255, 15, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] }) })) }], identity: "9cab9ecb-0b18-4956-a242-784c3a0b8dab", node_body: Some(Project(ProjectNode { select_list: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 0 })) }, ExprNode { expr_type: RoundDigit, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(FuncCall(FunctionCall { children: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 1 })) }, ExprNode { expr_type: ConstantValue, return_type: Some(DataType { type_name: Int32, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(Constant(ConstantValue { body: [0, 0, 0, 2] })) }] })) }, ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 2 })) }] })) }), exchange_info: Some(ExchangeInfo { mode: Single, distribution: None }) }), epoch: 3121804801998848 })) }, ExchangeSource { task_output_id: Some(TaskOutputId { task_id: Some(TaskId { query_id: "587e7934-8a8a-487c-8645-f78d39a0094f", stage_id: 1, task_id: 10 }), output_id: 0 }), host: Some(HostAddress { host: "127.0.0.1", port: 5689 }), local_execute_plan: Some(Plan(LocalExecutePlan { plan: Some(PlanFragment { root: Some(PlanNode { children: [PlanNode { children: [], identity: "e22e772b-1e96-43f2-97f8-53fab4da41f4", node_body: Some(RowSeqScan(RowSeqScanNode { table_desc: Some(StorageTableDesc { table_id: 1221, columns: [ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1000, name: "ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1001, name: "ctr", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1002, name: "window_end", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1003, name: "ad_impression.ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1004, name: "(TumbleStart(ad_impression.impression_timestamp, '00:01:00':Interval) + '00:01:00':Interval)", field_descs: [], type_name: "" }], pk: [ColumnOrder { order_type: Ascending, index: 3 }, ColumnOrder { order_type: Ascending, index: 2 }, ColumnOrder { order_type: Ascending, index: 0 }, ColumnOrder { order_type: Ascending, index: 4 }], dist_key_indices: [3, 2], retention_seconds: 0, value_indices: [0, 1, 2, 3, 4] }), column_ids: [1000, 1001, 1002], scan_ranges: [], vnode_bitmap: Some(Buffer { compression: None, body: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 254, 255, 63, 0, 0, 0, 0, 0] }) })) }], identity: "c62a77f7-b6e8-44f9-b95b-55f357b14185", node_body: Some(Project(ProjectNode { select_list: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 0 })) }, ExprNode { expr_type: RoundDigit, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(FuncCall(FunctionCall { children: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 1 })) }, ExprNode { expr_type: ConstantValue, return_type: Some(DataType { type_name: Int32, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(Constant(ConstantValue { body: [0, 0, 0, 2] })) }] })) }, ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 2 })) }] })) }), exchange_info: Some(ExchangeInfo { mode: Single, distribution: None }) }), epoch: 3121804801998848 })) }, ExchangeSource { task_output_id: Some(TaskOutputId { task_id: Some(TaskId { query_id: "587e7934-8a8a-487c-8645-f78d39a0094f", stage_id: 1, task_id: 11 }), output_id: 0 }), host: Some(HostAddress { host: "127.0.0.1", port: 5687 }), local_execute_plan: Some(Plan(LocalExecutePlan { plan: Some(PlanFragment { root: Some(PlanNode { children: [PlanNode { children: [], identity: "424d8a88-3e81-4b80-85de-96a0bb80f33d", node_body: Some(RowSeqScan(RowSeqScanNode { table_desc: Some(StorageTableDesc { table_id: 1221, columns: [ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1000, name: "ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1001, name: "ctr", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1002, name: "window_end", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1003, name: "ad_impression.ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1004, name: "(TumbleStart(ad_impression.impression_timestamp, '00:01:00':Interval) + '00:01:00':Interval)", field_descs: [], type_name: "" }], pk: [ColumnOrder { order_type: Ascending, index: 3 }, ColumnOrder { order_type: Ascending, index: 2 }, ColumnOrder { order_type: Ascending, index: 0 }, ColumnOrder { order_type: Ascending, index: 4 }], dist_key_indices: [3, 2], retention_seconds: 0, value_indices: [0, 1, 2, 3, 4] }), column_ids: [1000, 1001, 1002], scan_ranges: [], vnode_bitmap: Some(Buffer { compression: None, body: [0, 0, 0, 192, 255, 255, 15, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] }) })) }], identity: "647a1843-2796-4601-9d85-33f13964d2fe", node_body: Some(Project(ProjectNode { select_list: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 0 })) }, ExprNode { expr_type: RoundDigit, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(FuncCall(FunctionCall { children: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 1 })) }, ExprNode { expr_type: ConstantValue, return_type: Some(DataType { type_name: Int32, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(Constant(ConstantValue { body: [0, 0, 0, 2] })) }] })) }, ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 2 })) }] })) }), exchange_info: Some(ExchangeInfo { mode: Single, distribution: None }) }), epoch: 3121804801998848 })) }], input_schema: [Field { data_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), name: "ad_ctr_5min.ad_id" }, Field { data_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), name: "RoundDigit(ad_ctr_5min.ctr, 2:Int32)" }, Field { data_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), name: "ad_ctr_5min.window_end" }] }))] Failed to build executor: internal error: [PlanNode: Ok(Project(ProjectNode { select_list: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 0 })) }, ExprNode { expr_type: RoundDigit, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(FuncCall(FunctionCall { children: [ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 1 })) }, ExprNode { expr_type: ConstantValue, return_type: Some(DataType { type_name: Int32, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(Constant(ConstantValue { body: [0, 0, 0, 2] })) }] })) }, ExprNode { expr_type: InputRef, return_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), rex_node: Some(InputRef(InputRefExpr { column_idx: 2 })) }] }))] Failed to build executor: internal error: [PlanNode: Ok(RowSeqScan(RowSeqScanNode { table_desc: Some(StorageTableDesc { table_id: 1221, columns: [ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1000, name: "ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Decimal, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1001, name: "ctr", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1002, name: "window_end", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Int64, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1003, name: "ad_impression.ad_id", field_descs: [], type_name: "" }, ColumnDesc { column_type: Some(DataType { type_name: Timestamp, precision: 0, scale: 0, is_nullable: true, interval_type: Unspecified, field_type: [], field_names: [] }), column_id: 1004, name: "(TumbleStart(ad_impression.impression_timestamp, '00:01:00':Interval) + '00:01:00':Interval)", field_descs: [], type_name: "" }], pk: [ColumnOrder { order_type: Ascending, index: 3 }, ColumnOrder { order_type: Ascending, index: 2 }, ColumnOrder { order_type: Ascending, index: 0 }, ColumnOrder { order_type: Ascending, index: 4 }], dist_key_indices: [3, 2], retention_seconds: 0, value_indices: [0, 1, 2, 3, 4] }), column_ids: [1000, 1001, 1002], scan_ranges: [], vnode_bitmap: Some(Buffer { compression: None, body: [0, 255, 255, 63, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] }) }))] Failed to build executor: Storage error: Hummock error: Expired Epoch: watermark 3121804825198592, epoch 3121804801998848.   | [SQL] SELECT ad_id, ROUND(ctr, 2), window_end FROM ad_ctr_5min;   | at e2e_test/streaming/demo/ad_ctr.slt:104

logs
risedev-logs.zip

To Reproduce

No response

Expected behavior

No response

Additional context

No response

[BugenZhao]

The error is unreadable. 🤣

risingwave/src/batch/src/executor/mod.rs

Lines 165 to 174 in 7957c6f

	pub async fn build(&self) -> Result<BoxedExecutor> {
	self.try_build().await.map_err(|e| {
	anyhow!(format!(
	"[PlanNode: {:?}] Failed to build executor: {}",
	self.plan_node.get_node_body(),
	e,
	))
	.into()
	})
	}

[hzxa21]

frontend-4567.log:

2022-10-04T07:56:02.695058Z  INFO risingwave_frontend::scheduler::hummock_snapshot_manager: Unpin epoch 3121804825198592 with RPC

compute-node-5687

2022-10-04T07:56:02.984341Z ERROR risingwave_storage::monitor::monitored_store: Failed in iter: Hummock error: Expired Epoch: watermark 3121804825198592, epoch 3121804801998848.

frontend unpins epoch 3121804825198592 before CN finish contrusting the iterator with read epoch 3121804801998848.

3121804825198592 > 3121804801998848

[BugenZhao]

BTW, how about refactoring the snapshot ref count with Arc in the frontend (as what we do with PinnedVersion in compute node)?

[hzxa21]

BTW, how about refactoring the snapshot ref count with Arc in the frontend (as what we do with PinnedVersion in compute node)?

+1

[chenzl25]

https://buildkite.com/risingwavelabs/pull-request/builds/10601#0183f4b5-d85c-45a0-982a-9ef5a1d95449

frontend.log