Repository: pingcap/tidb (go)

Issue:
EXTRACT(YEAR FROM col) partitioning expression does not prune partitions
## Bug Report

Please answer these questions before submitting your issue. Thanks!

### 1. Minimal reproduce step (Required)

<!-- a step by step guide for reproducing the bug. -->
```mysql
create table t (a datetime, b varchar(255)) partition by range (extract(year from a)) (partition p2020 values less than (2021), partition p2021 values less than (2022), partition pMax values less than (maxvalue));
insert into t values ('2020-12-31 23:59:59','2020-12-31 23:59:59'),('2021-01-01', '2021-01-01');
explain select * from t where a = '2020-12-31 23:59:59';
```

### 2. What did you expect to see? (Required)
That only partition p2020 would be used.
```mysql
tidb> explain select * from t where a = '2020-12-31 23:59:59';
+-------------------------+----------+-----------+-----------------+------------------------------------------+
| id                      | estRows  | task      | access object   | operator info                            |
+-------------------------+----------+-----------+-----------------+------------------------------------------+
| TableReader_7           | 10.00    | root      | partition:p2020 | data:Selection_6                         |
| └─Selection_6       | 10.00    | cop[tikv] |                 | eq(test.t.a, 2020-12-31 23:59:59.000000) |
|   └─TableFullScan_5 | 10000.00 | cop[tikv] | table:t         | keep order:false, stats:pseudo           |
+-------------------------+----------+-----------+-----------------+------------------------------------------+

```
### 3. What did you see instead (Required)
All partitions where used.
```mysql
tidb> explain select * from t where a = '2020-12-31 23:59:59';
+-------------------------+----------+-----------+---------------+------------------------------------------+
| id                      | estRows  | task      | access object | operator info                            |
+-------------------------+----------+-----------+---------------+------------------------------------------+
| TableReader_7           | 10.00    | root      | partition:all | data:Selection_6                         |
| └─Selection_6       | 10.00    | cop[tikv] |               | eq(test.t.a, 2020-12-31 23:59:59.000000) |
|   └─TableFullScan_5 | 10000.00 | cop[tikv] | table:t       | keep order:false, stats:pseudo           |
+-------------------------+----------+-----------+---------------+------------------------------------------+
```
### 4. What is your TiDB version? (Required)

<!-- Paste the output of SELECT tidb_version() -->
```
tidb_version(): Release Version: v8.1.0
Edition: Community
Git Commit Hash: 945d07c5d5c7a1ae212f6013adfb187f2de24b23
Git Branch: HEAD
UTC Build Time: 2024-05-21 03:52:40
GoVersion: go1.21.10
Race Enabled: false
Check Table Before Drop: false
Store: tikv
```


Pull request description:
planner: Optimized EXTRACT(<UNIT> FROM <partition_column>) for partition pruning.
<!--

Thank you for contributing to TiDB!

PR Title Format:
1. pkg [, pkg2, pkg3]: what's changed
2. *: what's changed

-->

### What problem does this PR solve?
<!--

Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.

-->

Issue Number: close #54210

Problem Summary:
the expression `EXTRACT(<time unit> FROM col)` is not handled for partition pruning.

### What changed and how does it work?
Treat the expression as prunable for all time units except WEEK (due to `default_week_format`).
Meaning all prunable expressions will work with EQ/`col = <date/datetime/time>`.
For DATE/DATETIME columns, it also optimizes YEAR and YEAR_MONTH to also work with ranges (<, <=, >, >=, BETWEEN), and for TIME columns it also optimizes HOUR, HOUR_MINUTE, HOUR_SECOND and HOUR_MICROSECOND for ranges.


### Check List

Tests <!-- At least one of them must be included. -->

- [x] Unit test
- [ ] Integration test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test
  > - [ ] I checked and no code files have been changed.
  > <!-- Or your custom  "No need to test" reasons -->

Side effects

- [ ] Performance regression: Consumes more CPU
- [ ] Performance regression: Consumes more Memory
- [ ] Breaking backward compatibility

Documentation

- [ ] Affects user behaviors
- [ ] Contains syntax changes
- [ ] Contains variable changes
- [ ] Contains experimental features
- [ ] Changes MySQL compatibility

### Release note

<!-- compatibility change, improvement, bugfix, and new feature need a release note -->

Please refer to [Release Notes Language Style Guide](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/release-notes-style-guide.html) to write a quality release note.

```release-note
None
```


Patch:
diff --git a/pkg/executor/BUILD.bazel b/pkg/executor/BUILD.bazel
index ecf3c69cfe..0960da8dc2 100644
--- a/pkg/executor/BUILD.bazel
+++ b/pkg/executor/BUILD.bazel
@@ -412,6 +412,7 @@ go_test(
         "//pkg/planner",
         "//pkg/planner/core",
         "//pkg/planner/core/base",
+        "//pkg/planner/core/operator/logicalop",
         "//pkg/planner/property",
         "//pkg/planner/util",
         "//pkg/planner/util/coretestsdk",
diff --git a/pkg/executor/builder.go b/pkg/executor/builder.go
index d54046b2d0..6bcee9f5f8 100644
--- a/pkg/executor/builder.go
+++ b/pkg/executor/builder.go
@@ -62,6 +62,7 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/terror"
 	plannercore "github.com/pingcap/tidb/pkg/planner/core"
 	"github.com/pingcap/tidb/pkg/planner/core/base"
+	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
 	plannerutil "github.com/pingcap/tidb/pkg/planner/util"
 	"github.com/pingcap/tidb/pkg/planner/util/coreusage"
 	"github.com/pingcap/tidb/pkg/sessionctx"
@@ -4961,11 +4962,11 @@ func (b *executorBuilder) buildWindow(v *plannercore.PhysicalWindow) exec.Execut
 		exec.windowFuncs = windowFuncs
 		exec.partialResults = partialResults
 		if v.Frame == nil {
-			exec.start = &plannercore.FrameBound{
+			exec.start = &logicalop.FrameBound{
 				Type:      ast.Preceding,
 				UnBounded: true,
 			}
-			exec.end = &plannercore.FrameBound{
+			exec.end = &logicalop.FrameBound{
 				Type:      ast.Following,
 				UnBounded: true,
 			}
diff --git a/pkg/executor/pipelined_window.go b/pkg/executor/pipelined_window.go
index 9bd7ea5256..5f6bbf2699 100644
--- a/pkg/executor/pipelined_window.go
+++ b/pkg/executor/pipelined_window.go
@@ -23,7 +23,7 @@ import (
 	"github.com/pingcap/tidb/pkg/executor/internal/vecgroupchecker"
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/parser/ast"
-	"github.com/pingcap/tidb/pkg/planner/core"
+	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/util/chunk"
 )
@@ -41,8 +41,8 @@ type PipelinedWindowExec struct {
 	windowFuncs        []aggfuncs.AggFunc
 	slidingWindowFuncs []aggfuncs.SlidingWindowAggFunc
 	partialResults     []aggfuncs.PartialResult
-	start              *core.FrameBound
-	end                *core.FrameBound
+	start              *logicalop.FrameBound
+	end                *logicalop.FrameBound
 	groupChecker       *vecgroupchecker.VecGroupChecker
 
 	// childResult stores the child chunk. Note that even if remaining is 0, e.rows might still references rows in data[0].chk after returned it to upper executor, since there is no guarantee what the upper executor will do to the returned chunk, it might destroy the data (as in the benchmark test, it reused the chunk to pull data, and it will be chk.Reset(), causing panicking). So dataIdx, accumulated and dropped are added to ensure that chunk will only be returned if there is no row reference.
diff --git a/pkg/executor/window.go b/pkg/executor/window.go
index 42c46cba9f..4b7c7b68a0 100644
--- a/pkg/executor/window.go
+++ b/pkg/executor/window.go
@@ -23,7 +23,7 @@ import (
 	"github.com/pingcap/tidb/pkg/executor/internal/vecgroupchecker"
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/parser/ast"
-	"github.com/pingcap/tidb/pkg/planner/core"
+	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/util/chunk"
 )
@@ -237,8 +237,8 @@ func (p *aggWindowProcessor) resetPartialResult() {
 type rowFrameWindowProcessor struct {
 	windowFuncs    []aggfuncs.AggFunc
 	partialResults []aggfuncs.PartialResult
-	start          *core.FrameBound
-	end            *core.FrameBound
+	start          *logicalop.FrameBound
+	end            *logicalop.FrameBound
 	curRowIdx      uint64
 }
 
@@ -379,8 +379,8 @@ func (p *rowFrameWindowProcessor) resetPartialResult() {
 type rangeFrameWindowProcessor struct {
 	windowFuncs     []aggfuncs.AggFunc
 	partialResults  []aggfuncs.PartialResult
-	start           *core.FrameBound
-	end             *core.FrameBound
+	start           *logicalop.FrameBound
+	end             *logicalop.FrameBound
 	curRowIdx       uint64
 	lastStartOffset uint64
 	lastEndOffset   uint64
diff --git a/pkg/planner/cascades/implementation_rules.go b/pkg/planner/cascades/implementation_rules.go
index 409c036e6f..4abda4a43e 100644
--- a/pkg/planner/cascades/implementation_rules.go
+++ b/pkg/planner/cascades/implementation_rules.go
@@ -597,7 +597,7 @@ type ImplWindow struct {
 
 // Match implements ImplementationRule Match interface.
 func (*ImplWindow) Match(expr *memo.GroupExpr, prop *property.PhysicalProperty) (matched bool) {
-	lw := expr.ExprNode.(*plannercore.LogicalWindow)
+	lw := expr.ExprNode.(*logicalop.LogicalWindow)
 	var byItems []property.SortItem
 	byItems = append(byItems, lw.PartitionBy...)
 	byItems = append(byItems, lw.OrderBy...)
@@ -607,7 +607,7 @@ func (*ImplWindow) Match(expr *memo.GroupExpr, prop *property.PhysicalProperty)
 
 // OnImplement implements ImplementationRule OnImplement interface.
 func (*ImplWindow) OnImplement(expr *memo.GroupExpr, reqProp *property.PhysicalProperty) ([]memo.Implementation, error) {
-	lw := expr.ExprNode.(*plannercore.LogicalWindow)
+	lw := expr.ExprNode.(*logicalop.LogicalWindow)
 	var byItems []property.SortItem
 	byItems = append(byItems, lw.PartitionBy...)
 	byItems = append(byItems, lw.OrderBy...)
diff --git a/pkg/planner/cascades/transformation_rules.go b/pkg/planner/cascades/transformation_rules.go
index ff8addd759..911c8f2dad 100644
--- a/pkg/planner/cascades/transformation_rules.go
+++ b/pkg/planner/cascades/transformation_rules.go
@@ -684,7 +684,7 @@ func NewRulePushSelDownWindow() Transformation {
 // 3. just keep unchanged.
 func (*PushSelDownWindow) OnTransform(old *memo.ExprIter) (newExprs []*memo.GroupExpr, eraseOld bool, eraseAll bool, err error) {
 	sel := old.GetExpr().ExprNode.(*plannercore.LogicalSelection)
-	window := old.Children[0].GetExpr().ExprNode.(*plannercore.LogicalWindow)
+	window := old.Children[0].GetExpr().ExprNode.(*logicalop.LogicalWindow)
 	windowSchema := old.Children[0].Prop.Schema
 	childGroup := old.Children[0].GetExpr().Children[0]
 	canBePushed := make([]expression.Expression, 0, len(sel.Conditions))
@@ -2542,9 +2542,9 @@ func NewRuleMergeAdjacentWindow() Transformation {
 
 // Match implements Transformation interface.
 func (*MergeAdjacentWindow) Match(expr *memo.ExprIter) bool {
-	curWinPlan := expr.GetExpr().ExprNode.(*plannercore.LogicalWindow)
+	curWinPlan := expr.GetExpr().ExprNode.(*logicalop.LogicalWindow)
 	nextGroupExpr := expr.Children[0].GetExpr()
-	nextWinPlan := nextGroupExpr.ExprNode.(*plannercore.LogicalWindow)
+	nextWinPlan := nextGroupExpr.ExprNode.(*logicalop.LogicalWindow)
 	nextGroupChildren := nextGroupExpr.Children
 	ctx := expr.GetExpr().ExprNode.SCtx()
 
@@ -2582,14 +2582,14 @@ func (*MergeAdjacentWindow) Match(expr *memo.ExprIter) bool {
 // OnTransform implements Transformation interface.
 // This rule will transform `window -> window -> x` to `window -> x`
 func (*MergeAdjacentWindow) OnTransform(old *memo.ExprIter) (newExprs []*memo.GroupExpr, eraseOld bool, eraseAll bool, err error) {
-	curWinPlan := old.GetExpr().ExprNode.(*plannercore.LogicalWindow)
-	nextWinPlan := old.Children[0].GetExpr().ExprNode.(*plannercore.LogicalWindow)
+	curWinPlan := old.GetExpr().ExprNode.(*logicalop.LogicalWindow)
+	nextWinPlan := old.Children[0].GetExpr().ExprNode.(*logicalop.LogicalWindow)
 	ctx := old.GetExpr().ExprNode.SCtx()
 
 	newWindowFuncs := make([]*aggregation.WindowFuncDesc, 0, len(curWinPlan.WindowFuncDescs)+len(nextWinPlan.WindowFuncDescs))
 	newWindowFuncs = append(newWindowFuncs, curWinPlan.WindowFuncDescs...)
 	newWindowFuncs = append(newWindowFuncs, nextWinPlan.WindowFuncDescs...)
-	newWindowPlan := plannercore.LogicalWindow{
+	newWindowPlan := logicalop.LogicalWindow{
 		WindowFuncDescs: newWindowFuncs,
 		PartitionBy:     curWinPlan.PartitionBy,
 		OrderBy:         curWinPlan.OrderBy,
diff --git a/pkg/planner/core/BUILD.bazel b/pkg/planner/core/BUILD.bazel
index e48babee0a..26729d6276 100644
--- a/pkg/planner/core/BUILD.bazel
+++ b/pkg/planner/core/BUILD.bazel
@@ -39,7 +39,6 @@ go_library(
         "logical_tikv_single_gather.go",
         "logical_union_all.go",
         "logical_union_scan.go",
-        "logical_window.go",
         "memtable_predicate_extractor.go",
         "mock.go",
         "optimizer.go",
diff --git a/pkg/planner/core/collect_column_stats_usage.go b/pkg/planner/core/collect_column_stats_usage.go
index dcc837e6cd..54c427c822 100644
--- a/pkg/planner/core/collect_column_stats_usage.go
+++ b/pkg/planner/core/collect_column_stats_usage.go
@@ -258,7 +258,7 @@ func (c *columnStatsUsageCollector) collectFromPlan(lp base.LogicalPlan) {
 			for i, aggFunc := range x.AggFuncs {
 				c.updateColMapFromExpressions(schema.Columns[i], aggFunc.Args)
 			}
-		case *LogicalWindow:
+		case *logicalop.LogicalWindow:
 			// Statistics of the columns in LogicalWindow.PartitionBy are used in optimizeByShuffle4Window.
 			// We don't use statistics of the columns in LogicalWindow.OrderBy currently.
 			for _, item := range x.PartitionBy {
diff --git a/pkg/planner/core/core_init.go b/pkg/planner/core/core_init.go
index 7d12b7e6fb..128792b9b4 100644
--- a/pkg/planner/core/core_init.go
+++ b/pkg/planner/core/core_init.go
@@ -43,6 +43,7 @@ func init() {
 	utilfuncp.ExhaustPhysicalPlans4LogicalSort = exhaustPhysicalPlans4LogicalSort
 	utilfuncp.ExhaustPhysicalPlans4LogicalTopN = exhaustPhysicalPlans4LogicalTopN
 	utilfuncp.ExhaustPhysicalPlans4LogicalLimit = exhaustPhysicalPlans4LogicalLimit
+	utilfuncp.ExhaustLogicalWindowPhysicalPlans = exhaustLogicalWindowPhysicalPlans
 	utilfuncp.ExhaustPhysicalPlans4LogicalSequence = exhaustPhysicalPlans4LogicalSequence
 	utilfuncp.ExhaustPhysicalPlans4LogicalMaxOneRow = exhaustPhysicalPlans4LogicalMaxOneRow
 	utilfuncp.ExhaustPhysicalPlans4LogicalProjection = exhaustPhysicalPlans4LogicalProjection
diff --git a/pkg/planner/core/exhaust_physical_plans.go b/pkg/planner/core/exhaust_physical_plans.go
index 0dd3be74bd..977f396eb4 100644
--- a/pkg/planner/core/exhaust_physical_plans.go
+++ b/pkg/planner/core/exhaust_physical_plans.go
@@ -2183,7 +2183,7 @@ func disableAggPushDownToCop(p base.LogicalPlan) {
 	}
 }
 
-func tryToGetMppWindows(lw *LogicalWindow, prop *property.PhysicalProperty) []base.PhysicalPlan {
+func tryToGetMppWindows(lw *logicalop.LogicalWindow, prop *property.PhysicalProperty) []base.PhysicalPlan {
 	if !prop.IsSortItemAllForPartition() {
 		return nil
 	}
@@ -2280,7 +2280,8 @@ func tryToGetMppWindows(lw *LogicalWindow, prop *property.PhysicalProperty) []ba
 	return []base.PhysicalPlan{window}
 }
 
-func exhaustLogicalWindowPhysicalPlans(lw *LogicalWindow, prop *property.PhysicalProperty) ([]base.PhysicalPlan, bool, error) {
+func exhaustLogicalWindowPhysicalPlans(lp base.LogicalPlan, prop *property.PhysicalProperty) ([]base.PhysicalPlan, bool, error) {
+	lw := lp.(*logicalop.LogicalWindow)
 	windows := make([]base.PhysicalPlan, 0, 2)
 
 	canPushToTiFlash := lw.CanPushToCop(kv.TiFlash)
@@ -2366,7 +2367,7 @@ func canPushToCopImpl(lp base.LogicalPlan, storeTp kv.StoreType, considerDual bo
 			ret = ret && canPushToCopImpl(&c.BaseLogicalPlan, storeTp, considerDual)
 		case *logicalop.LogicalTableDual:
 			return storeTp == kv.TiFlash && considerDual
-		case *LogicalAggregation, *LogicalSelection, *LogicalJoin, *LogicalWindow:
+		case *LogicalAggregation, *LogicalSelection, *LogicalJoin, *logicalop.LogicalWindow:
 			if storeTp != kv.TiFlash {
 				return false
 			}
diff --git a/pkg/planner/core/explain.go b/pkg/planner/core/explain.go
index 11b20120b5..63e294c5d6 100644
--- a/pkg/planner/core/explain.go
+++ b/pkg/planner/core/explain.go
@@ -26,6 +26,7 @@ import (
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
+	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
 	"github.com/pingcap/tidb/pkg/planner/property"
 	"github.com/pingcap/tidb/pkg/planner/util"
 	"github.com/pingcap/tidb/pkg/sessionctx/variable"
@@ -783,7 +784,7 @@ func (p *PhysicalTopN) ExplainNormalizedInfo() string {
 	return buffer.String()
 }
 
-func (p *PhysicalWindow) formatFrameBound(buffer *bytes.Buffer, bound *FrameBound) {
+func (p *PhysicalWindow) formatFrameBound(buffer *bytes.Buffer, bound *logicalop.FrameBound) {
 	if bound.Type == ast.CurrentRow {
 		buffer.WriteString("current row")
 		return
diff --git a/pkg/planner/core/logical_plan_builder.go b/pkg/planner/core/logical_plan_builder.go
index 267937af16..3191de02bd 100644
--- a/pkg/planner/core/logical_plan_builder.go
+++ b/pkg/planner/core/logical_plan_builder.go
@@ -6097,9 +6097,9 @@ func (b *PlanBuilder) buildByItemsForWindow(
 // buildWindowFunctionFrameBound builds the bounds of window function frames.
 // For type `Rows`, the bound expr must be an unsigned integer.
 // For type `Range`, the bound expr must be temporal or numeric types.
-func (b *PlanBuilder) buildWindowFunctionFrameBound(_ context.Context, spec *ast.WindowSpec, orderByItems []property.SortItem, boundClause *ast.FrameBound) (*FrameBound, error) {
+func (b *PlanBuilder) buildWindowFunctionFrameBound(_ context.Context, spec *ast.WindowSpec, orderByItems []property.SortItem, boundClause *ast.FrameBound) (*logicalop.FrameBound, error) {
 	frameType := spec.Frame.Type
-	bound := &FrameBound{Type: boundClause.Type, UnBounded: boundClause.UnBounded, IsExplicitRange: false}
+	bound := &logicalop.FrameBound{Type: boundClause.Type, UnBounded: boundClause.UnBounded, IsExplicitRange: false}
 	if bound.UnBounded {
 		return bound, nil
 	}
@@ -6184,18 +6184,18 @@ func (b *PlanBuilder) buildWindowFunctionFrameBound(_ context.Context, spec *ast
 	}
 
 	cmpDataType := expression.GetAccurateCmpType(b.ctx.GetExprCtx().GetEvalCtx(), col, bound.CalcFuncs[0])
-	bound.updateCmpFuncsAndCmpDataType(cmpDataType)
+	bound.UpdateCmpFuncsAndCmpDataType(cmpDataType)
 	return bound, nil
 }
 
 // buildWindowFunctionFrame builds the window function frames.
 // See https://dev.mysql.com/doc/refman/8.0/en/window-functions-frames.html
-func (b *PlanBuilder) buildWindowFunctionFrame(ctx context.Context, spec *ast.WindowSpec, orderByItems []property.SortItem) (*WindowFrame, error) {
+func (b *PlanBuilder) buildWindowFunctionFrame(ctx context.Context, spec *ast.WindowSpec, orderByItems []property.SortItem) (*logicalop.WindowFrame, error) {
 	frameClause := spec.Frame
 	if frameClause == nil {
 		return nil, nil
 	}
-	frame := &WindowFrame{Type: frameClause.Type}
+	frame := &logicalop.WindowFrame{Type: frameClause.Type}
 	var err error
 	frame.Start, err = b.buildWindowFunctionFrameBound(ctx, spec, orderByItems, &frameClause.Extent.Start)
 	if err != nil {
@@ -6322,7 +6322,7 @@ func (b *PlanBuilder) buildWindowFunctions(ctx context.Context, p base.LogicalPl
 			return nil, nil, err
 		}
 
-		window := LogicalWindow{
+		window := logicalop.LogicalWindow{
 			PartitionBy: partitionBy,
 			OrderBy:     orderBy,
 			Frame:       frame,
diff --git a/pkg/planner/core/logical_plans.go b/pkg/planner/core/logical_plans.go
index 738796dabb..861732b353 100644
--- a/pkg/planner/core/logical_plans.go
+++ b/pkg/planner/core/logical_plans.go
@@ -39,7 +39,7 @@ var (
 	_ base.LogicalPlan = &logicalop.LogicalSort{}
 	_ base.LogicalPlan = &LogicalLock{}
 	_ base.LogicalPlan = &logicalop.LogicalLimit{}
-	_ base.LogicalPlan = &LogicalWindow{}
+	_ base.LogicalPlan = &logicalop.LogicalWindow{}
 	_ base.LogicalPlan = &LogicalExpand{}
 	_ base.LogicalPlan = &LogicalUnionScan{}
 	_ base.LogicalPlan = &logicalop.LogicalMemTable{}
diff --git a/pkg/planner/core/logical_selection.go b/pkg/planner/core/logical_selection.go
index 1541709f1a..c5f0883bf5 100644
--- a/pkg/planner/core/logical_selection.go
+++ b/pkg/planner/core/logical_selection.go
@@ -162,7 +162,7 @@ func (p *LogicalSelection) DeriveTopN(opt *optimizetrace.LogicalOptimizeOp) base
 	s := p.Self().(*LogicalSelection)
 	windowIsTopN, limitValue := windowIsTopN(s)
 	if windowIsTopN {
-		child := s.Children()[0].(*LogicalWindow)
+		child := s.Children()[0].(*logicalop.LogicalWindow)
 		grandChild := child.Children()[0].(*DataSource)
 		// Build order by for derived Limit
 		byItems := make([]*util.ByItems, 0, len(child.OrderBy))
diff --git a/pkg/planner/core/operator/logicalop/BUILD.bazel b/pkg/planner/core/operator/logicalop/BUILD.bazel
index 5786c9cd31..6690041405 100644
--- a/pkg/planner/core/operator/logicalop/BUILD.bazel
+++ b/pkg/planner/core/operator/logicalop/BUILD.bazel
@@ -16,11 +16,13 @@ go_library(
         "logical_sort.go",
         "logical_table_dual.go",
         "logical_top_n.go",
+        "logical_window.go",
     ],
     importpath = "github.com/pingcap/tidb/pkg/planner/core/operator/logicalop",
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/expression",
+        "//pkg/expression/aggregation",
         "//pkg/infoschema",
         "//pkg/kv",
         "//pkg/parser/ast",
@@ -37,6 +39,7 @@ go_library(
         "//pkg/planner/util/optimizetrace",
         "//pkg/planner/util/optimizetrace/logicaltrace",
         "//pkg/planner/util/utilfuncp",
+        "//pkg/sessionctx",
         "//pkg/statistics",
         "//pkg/types",
         "//pkg/util/dbterror/plannererrors",
@@ -45,5 +48,6 @@ go_library(
         "//pkg/util/size",
         "//pkg/util/tracing",
         "@com_github_pingcap_errors//:errors",
+        "@com_github_pingcap_tipb//go-tipb",
     ],
 )
diff --git a/pkg/planner/core/logical_window.go b/pkg/planner/core/operator/logicalop/logical_window.go
similarity index 94%
rename from pkg/planner/core/logical_window.go
rename to pkg/planner/core/operator/logicalop/logical_window.go
index 6f958dcf8e..71e0d5cac6 100644
--- a/pkg/planner/core/logical_window.go
+++ b/pkg/planner/core/operator/logicalop/logical_window.go
@@ -12,17 +12,17 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package core
+package logicalop
 
 import (
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/expression/aggregation"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/pingcap/tidb/pkg/planner/core/base"
-	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
 	ruleutil "github.com/pingcap/tidb/pkg/planner/core/rule/util"
 	"github.com/pingcap/tidb/pkg/planner/property"
 	"github.com/pingcap/tidb/pkg/planner/util/optimizetrace"
+	"github.com/pingcap/tidb/pkg/planner/util/utilfuncp"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util/plancodec"
@@ -31,7 +31,7 @@ import (
 
 // LogicalWindow represents a logical window function plan.
 type LogicalWindow struct {
-	logicalop.LogicalSchemaProducer
+	LogicalSchemaProducer
 
 	WindowFuncDescs []*aggregation.WindowFuncDesc
 	PartitionBy     []property.SortItem
@@ -90,7 +90,8 @@ func (fb *FrameBound) Clone() *FrameBound {
 	return cloned
 }
 
-func (fb *FrameBound) updateCmpFuncsAndCmpDataType(cmpDataType types.EvalType) {
+// UpdateCmpFuncsAndCmpDataType updates CmpFuncs and CmpDataType.
+func (fb *FrameBound) UpdateCmpFuncsAndCmpDataType(cmpDataType types.EvalType) {
 	// When cmpDataType can't match to any condition, we can ignore it.
 	//
 	// For example:
@@ -117,6 +118,28 @@ func (fb *FrameBound) updateCmpFuncsAndCmpDataType(cmpDataType types.EvalType) {
 	}
 }
 
+// ToPB converts FrameBound to tipb structure.
+func (fb *FrameBound) ToPB(ctx *base.BuildPBContext) (*tipb.WindowFrameBound, error) {
+	pbBound := &tipb.WindowFrameBound{
+		Type:      tipb.WindowBoundType(fb.Type),
+		Unbounded: fb.UnBounded,
+	}
+	offset := fb.Num
+	pbBound.Offset = &offset
+
+	if fb.IsExplicitRange {
+		rangeFrame, err := expression.ExpressionsToPBList(ctx.GetExprCtx().GetEvalCtx(), fb.CalcFuncs, ctx.GetClient())
+		if err != nil {
+			return nil, err
+		}
+
+		pbBound.FrameRange = rangeFrame[0]
+		pbBound.CmpDataType = &fb.CmpDataType
+	}
+
+	return pbBound, nil
+}
+
 // UpdateCompareCols will update CompareCols.
 func (fb *FrameBound) UpdateCompareCols(ctx sessionctx.Context, orderByCols []*expression.Column) error {
 	ectx := ctx.GetExprCtx().GetEvalCtx()
@@ -136,14 +159,14 @@ func (fb *FrameBound) UpdateCompareCols(ctx sessionctx.Context, orderByCols []*e
 		}
 
 		cmpDataType := expression.GetAccurateCmpType(ctx.GetExprCtx().GetEvalCtx(), fb.CompareCols[0], fb.CalcFuncs[0])
-		fb.updateCmpFuncsAndCmpDataType(cmpDataType)
+		fb.UpdateCmpFuncsAndCmpDataType(cmpDataType)
 	}
 	return nil
 }
 
 // Init initializes LogicalWindow.
 func (p LogicalWindow) Init(ctx base.PlanContext, offset int) *LogicalWindow {
-	p.BaseLogicalPlan = logicalop.NewBaseLogicalPlan(ctx, plancodec.TypeWindow, &p, offset)
+	p.BaseLogicalPlan = NewBaseLogicalPlan(ctx, plancodec.TypeWindow, &p, offset)
 	return &p
 }
 
@@ -289,7 +312,7 @@ func (p *LogicalWindow) PreparePossibleProperties(_ *expression.Schema, _ ...[][
 
 // ExhaustPhysicalPlans implements base.LogicalPlan.<14th> interface.
 func (p *LogicalWindow) ExhaustPhysicalPlans(prop *property.PhysicalProperty) ([]base.PhysicalPlan, bool, error) {
-	return exhaustLogicalWindowPhysicalPlans(p, prop)
+	return utilfuncp.ExhaustLogicalWindowPhysicalPlans(p, prop)
 }
 
 // ExtractCorrelatedCols implements base.LogicalPlan.<15th> interface.
diff --git a/pkg/planner/core/physical_plans.go b/pkg/planner/core/physical_plans.go
index 7af1609ce3..f13916846d 100644
--- a/pkg/planner/core/physical_plans.go
+++ b/pkg/planner/core/physical_plans.go
@@ -2392,7 +2392,7 @@ type PhysicalWindow struct {
 	WindowFuncDescs []*aggregation.WindowFuncDesc
 	PartitionBy     []property.SortItem
 	OrderBy         []property.SortItem
-	Frame           *WindowFrame
+	Frame           *logicalop.WindowFrame
 
 	// on which store the window function executes.
 	storeTp kv.StoreType
diff --git a/pkg/planner/core/plan.go b/pkg/planner/core/plan.go
index 647ac91102..a923268e03 100644
--- a/pkg/planner/core/plan.go
+++ b/pkg/planner/core/plan.go
@@ -384,7 +384,7 @@ func HasMaxOneRow(p base.LogicalPlan, childMaxOneRow []bool) bool {
 	}
 	switch x := p.(type) {
 	case *LogicalLock, *logicalop.LogicalLimit, *logicalop.LogicalSort, *LogicalSelection,
-		*LogicalApply, *logicalop.LogicalProjection, *LogicalWindow, *LogicalAggregation:
+		*LogicalApply, *logicalop.LogicalProjection, *logicalop.LogicalWindow, *LogicalAggregation:
 		return childMaxOneRow[0]
 	case *logicalop.LogicalMaxOneRow:
 		return true
diff --git a/pkg/planner/core/plan_to_pb.go b/pkg/planner/core/plan_to_pb.go
index 970f13058e..1cc9073289 100644
--- a/pkg/planner/core/plan_to_pb.go
+++ b/pkg/planner/core/plan_to_pb.go
@@ -636,28 +636,6 @@ func (p *PhysicalHashJoin) ToPB(ctx *base.BuildPBContext, storeType kv.StoreType
 	}, nil
 }
 
-// ToPB converts FrameBound to tipb structure.
-func (fb *FrameBound) ToPB(ctx *base.BuildPBContext) (*tipb.WindowFrameBound, error) {
-	pbBound := &tipb.WindowFrameBound{
-		Type:      tipb.WindowBoundType(fb.Type),
-		Unbounded: fb.UnBounded,
-	}
-	offset := fb.Num
-	pbBound.Offset = &offset
-
-	if fb.IsExplicitRange {
-		rangeFrame, err := expression.ExpressionsToPBList(ctx.GetExprCtx().GetEvalCtx(), fb.CalcFuncs, ctx.GetClient())
-		if err != nil {
-			return nil, err
-		}
-
-		pbBound.FrameRange = rangeFrame[0]
-		pbBound.CmpDataType = &fb.CmpDataType
-	}
-
-	return pbBound, nil
-}
-
 // ToPB implements PhysicalPlan ToPB interface.
 func (p *PhysicalWindow) ToPB(ctx *base.BuildPBContext, storeType kv.StoreType) (*tipb.Executor, error) {
 	client := ctx.GetClient()
diff --git a/pkg/planner/core/rule_derive_topn_from_window.go b/pkg/planner/core/rule_derive_topn_from_window.go
index 5b0ce8e4de..46712ed142 100644
--- a/pkg/planner/core/rule_derive_topn_from_window.go
+++ b/pkg/planner/core/rule_derive_topn_from_window.go
@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/pingcap/tidb/pkg/planner/core/base"
+	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
 	"github.com/pingcap/tidb/pkg/planner/util/optimizetrace"
 )
 
@@ -42,7 +43,7 @@ func appendDerivedTopNTrace(topN base.LogicalPlan, opt *optimizetrace.LogicalOpt
 
 // checkPartitionBy mainly checks if partition by of window function is a prefix of
 // data order (clustered index) of the data source. TiFlash is allowed only for empty partition by.
-func checkPartitionBy(p *LogicalWindow, d *DataSource) bool {
+func checkPartitionBy(p *logicalop.LogicalWindow, d *DataSource) bool {
 	// No window partition by. We are OK.
 	if len(p.PartitionBy) == 0 {
 		return true
@@ -75,7 +76,7 @@ func checkPartitionBy(p *LogicalWindow, d *DataSource) bool {
 */
 func windowIsTopN(p *LogicalSelection) (bool, uint64) {
 	// Check if child is window function.
-	child, isLogicalWindow := p.Children()[0].(*LogicalWindow)
+	child, isLogicalWindow := p.Children()[0].(*logicalop.LogicalWindow)
 	if !isLogicalWindow {
 		return false, 0
 	}
diff --git a/pkg/planner/core/rule_eliminate_projection.go b/pkg/planner/core/rule_eliminate_projection.go
index f05e3bea38..c323fb20cd 100644
--- a/pkg/planner/core/rule_eliminate_projection.go
+++ b/pkg/planner/core/rule_eliminate_projection.go
@@ -169,7 +169,7 @@ func (pe *projectionEliminator) eliminate(p base.LogicalPlan, replace map[string
 		childFlag = false
 	} else if _, isAgg := p.(*LogicalAggregation); isAgg || isProj {
 		childFlag = true
-	} else if _, isWindow := p.(*LogicalWindow); isWindow {
+	} else if _, isWindow := p.(*logicalop.LogicalWindow); isWindow {
 		childFlag = true
 	}
 	for i, child := range p.Children() {
diff --git a/pkg/planner/core/rule_partition_processor.go b/pkg/planner/core/rule_partition_processor.go
index c066ad9751..7b276c32ae 100644
--- a/pkg/planner/core/rule_partition_processor.go
+++ b/pkg/planner/core/rule_partition_processor.go
@@ -42,6 +42,7 @@ import (
 	"github.com/pingcap/tidb/pkg/util/plancodec"
 	"github.com/pingcap/tidb/pkg/util/ranger"
 	"github.com/pingcap/tidb/pkg/util/set"
+	"github.com/pingcap/tipb/go-tipb"
 )
 
 // FullRange represent used all partitions.
@@ -1067,9 +1068,11 @@ func makePartitionByFnCol(sctx base.PlanContext, columns []*expression.Column, n
 	switch raw := partExpr.(type) {
 	case *expression.ScalarFunction:
 		args := raw.GetArgs()
-		// Special handle for floor(unix_timestamp(ts)) as partition expression.
-		// This pattern is so common for timestamp(3) column as partition expression that it deserve an optimization.
-		if raw.FuncName.L == ast.Floor {
+		// Optimizations for a limited set of functions
+		switch raw.FuncName.L {
+		case ast.Floor:
+			// Special handle for floor(unix_timestamp(ts)) as partition expression.
+			// This pattern is so common for timestamp(3) column as partition expression that it deserve an optimization.
 			if ut, ok := args[0].(*expression.ScalarFunction); ok && ut.FuncName.L == ast.UnixTimestamp {
 				args1 := ut.GetArgs()
 				if len(args1) == 1 {
@@ -1078,6 +1081,58 @@ func makePartitionByFnCol(sctx base.PlanContext, columns []*expression.Column, n
 					}
 				}
 			}
+		case ast.Extract:
+			con, ok := args[0].(*expression.Constant)
+			if !ok {
+				break
+			}
+			col, ok = args[1].(*expression.Column)
+			if !ok {
+				// Special case where CastTimeToDuration is added
+				expr, ok := args[1].(*expression.ScalarFunction)
+				if !ok {
+					break
+				}
+				if expr.Function.PbCode() != tipb.ScalarFuncSig_CastTimeAsDuration {
+					break
+				}
+				castArgs := expr.GetArgs()
+				col, ok = castArgs[0].(*expression.Column)
+				if !ok {
+					break
+				}
+			}
+			if con.Value.Kind() != types.KindString {
+				break
+			}
+			val := con.Value.GetString()
+			colType := col.GetStaticType().GetType()
+			switch colType {
+			case mysql.TypeDate, mysql.TypeDatetime:
+				switch val {
+				// Only YEAR, YEAR_MONTH can be considered monotonic, the rest will wrap around!
+				case "YEAR", "YEAR_MONTH":
+					// Note, this function will not have the column as first argument,
+					// so in replaceColumnWithConst it will replace the second argument, which
+					// is special handling there too!
+					return col, raw, monotoneModeNonStrict, nil
+				default:
+					return col, raw, monotonous, nil
+				}
+			case mysql.TypeDuration:
+				switch val {
+				// Only HOUR* can be considered monotonic, the rest will wrap around!
+				// TODO: if fsp match for HOUR_SECOND or HOUR_MICROSECOND we could
+				// mark it as monotoneModeStrict
+				case "HOUR", "HOUR_MINUTE", "HOUR_SECOND", "HOUR_MICROSECOND":
+					// Note, this function will not have the column as first argument,
+					// so in replaceColumnWithConst it will replace the second argument, which
+					// is special handling there too!
+					return col, raw, monotoneModeNonStrict, nil
+				default:
+					return col, raw, monotonous, nil
+				}
+			}
 		}
 
 		fn = raw
@@ -1552,6 +1607,15 @@ func replaceColumnWithConst(partFn *expression.ScalarFunction, con *expression.C
 			args[0] = con
 			return partFn
 		}
+	} else if partFn.FuncName.L == ast.Extract {
+		if expr, ok := args[1].(*expression.ScalarFunction); ok && expr.Function.PbCode() == tipb.ScalarFuncSig_CastTimeAsDuration {
+			// Special handing if Cast is added
+			funcArgs := expr.GetArgs()
+			funcArgs[0] = con
+			return partFn
+		}
+		args[1] = con
+		return partFn
 	}
 
 	// No 'copy on write' for the expression here, this is a dangerous operation.
diff --git a/pkg/planner/core/stringer.go b/pkg/planner/core/stringer.go
index 00690241c3..4b98065c42 100644
--- a/pkg/planner/core/stringer.go
+++ b/pkg/planner/core/stringer.go
@@ -336,7 +336,7 @@ func toString(in base.Plan, strs []string, idxs []int) ([]string, []int) {
 		if x.SelectPlan != nil {
 			str = fmt.Sprintf("%s->Insert", ToString(x.SelectPlan))
 		}
-	case *LogicalWindow:
+	case *logicalop.LogicalWindow:
 		buffer := bytes.NewBufferString("")
 		formatWindowFuncDescs(ectx, buffer, x.WindowFuncDescs, x.Schema())
 		str = fmt.Sprintf("Window(%s)", buffer.String())
diff --git a/pkg/planner/pattern/pattern.go b/pkg/planner/pattern/pattern.go
index 6c525b029b..2b78e95fb2 100644
--- a/pkg/planner/pattern/pattern.go
+++ b/pkg/planner/pattern/pattern.go
@@ -115,7 +115,7 @@ func GetOperand(p base.LogicalPlan) Operand {
 		return OperandIndexScan
 	case *logicalop.LogicalShow:
 		return OperandShow
-	case *plannercore.LogicalWindow:
+	case *logicalop.LogicalWindow:
 		return OperandWindow
 	default:
 		return OperandUnsupported
diff --git a/pkg/planner/util/utilfuncp/func_pointer_misc.go b/pkg/planner/util/utilfuncp/func_pointer_misc.go
index fa16b43be2..d2b2ef7740 100644
--- a/pkg/planner/util/utilfuncp/func_pointer_misc.go
+++ b/pkg/planner/util/utilfuncp/func_pointer_misc.go
@@ -133,3 +133,7 @@ var ExhaustPhysicalPlans4LogicalLimit func(lp base.LogicalPlan, prop *property.P
 // ExhaustPhysicalPlans4LogicalProjection will be called by LogicalLimit in logicalOp pkg.
 var ExhaustPhysicalPlans4LogicalProjection func(lp base.LogicalPlan,
 	prop *property.PhysicalProperty) ([]base.PhysicalPlan, bool, error)
+
+// ExhaustLogicalWindowPhysicalPlans will be called by LogicalWindow in logicalOp pkg.
+var ExhaustLogicalWindowPhysicalPlans func(lp base.LogicalPlan, prop *property.PhysicalProperty) (
+	[]base.PhysicalPlan, bool, error)


Test patch:
diff --git a/pkg/executor/benchmark_test.go b/pkg/executor/benchmark_test.go
index 5d15670d25..2f91a32e8a 100644
--- a/pkg/executor/benchmark_test.go
+++ b/pkg/executor/benchmark_test.go
@@ -40,6 +40,7 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/planner/core"
 	"github.com/pingcap/tidb/pkg/planner/core/base"
+	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
 	"github.com/pingcap/tidb/pkg/planner/property"
 	"github.com/pingcap/tidb/pkg/planner/util"
 	"github.com/pingcap/tidb/pkg/sessionctx"
@@ -278,7 +279,7 @@ func BenchmarkAggDistinct(b *testing.B) {
 	}
 }
 
-func buildWindowExecutor(ctx sessionctx.Context, windowFunc string, funcs int, frame *core.WindowFrame, srcExec exec.Executor, schema *expression.Schema, partitionBy []*expression.Column, concurrency int, dataSourceSorted bool) exec.Executor {
+func buildWindowExecutor(ctx sessionctx.Context, windowFunc string, funcs int, frame *logicalop.WindowFrame, srcExec exec.Executor, schema *expression.Schema, partitionBy []*expression.Column, concurrency int, dataSourceSorted bool) exec.Executor {
 	src := testutil.BuildMockDataPhysicalPlan(ctx, srcExec)
 	win := new(core.PhysicalWindow)
 	win.WindowFuncDescs = make([]*aggregation.WindowFuncDesc, 0)
@@ -476,8 +477,8 @@ func baseBenchmarkWindowFunctionsWithFrame(b *testing.B, pipelined int) {
 		ast.AggFuncBitXor,
 	}
 	numFuncs := []int{1, 5}
-	frames := []*core.WindowFrame{
-		{Type: ast.Rows, Start: &core.FrameBound{UnBounded: true}, End: &core.FrameBound{Type: ast.CurrentRow}},
+	frames := []*logicalop.WindowFrame{
+		{Type: ast.Rows, Start: &logicalop.FrameBound{UnBounded: true}, End: &logicalop.FrameBound{Type: ast.CurrentRow}},
 	}
 	sortTypes := []bool{false, true}
 	concs := []int{1, 2, 3, 4, 5, 6}
@@ -513,7 +514,7 @@ func BenchmarkWindowFunctionsWithFrame(b *testing.B) {
 func baseBenchmarkWindowFunctionsAggWindowProcessorAboutFrame(b *testing.B, pipelined int) {
 	b.ReportAllocs()
 	windowFunc := ast.AggFuncMax
-	frame := &core.WindowFrame{Type: ast.Rows, Start: &core.FrameBound{UnBounded: true}, End: &core.FrameBound{UnBounded: true}}
+	frame := &logicalop.WindowFrame{Type: ast.Rows, Start: &logicalop.FrameBound{UnBounded: true}, End: &logicalop.FrameBound{UnBounded: true}}
 	cas := testutil.DefaultWindowTestCase()
 	cas.Rows = 10000
 	cas.Ndv = 10
@@ -552,10 +553,10 @@ func baseBenchmarkWindowFunctionsWithSlidingWindow(b *testing.B, frameType ast.F
 	}
 	row := 100000
 	ndv := 100
-	frame := &core.WindowFrame{
+	frame := &logicalop.WindowFrame{
 		Type:  frameType,
-		Start: &core.FrameBound{Type: ast.Preceding, Num: 10},
-		End:   &core.FrameBound{Type: ast.Following, Num: 10},
+		Start: &logicalop.FrameBound{Type: ast.Preceding, Num: 10},
+		End:   &logicalop.FrameBound{Type: ast.Following, Num: 10},
 	}
 	for _, windowFunc := range windowFuncs {
 		cas := testutil.DefaultWindowTestCase()
diff --git a/pkg/executor/internal/testutil/BUILD.bazel b/pkg/executor/internal/testutil/BUILD.bazel
index 40d5d3d0a1..4f1b73b227 100644
--- a/pkg/executor/internal/testutil/BUILD.bazel
+++ b/pkg/executor/internal/testutil/BUILD.bazel
@@ -16,8 +16,8 @@ go_library(
         "//pkg/expression",
         "//pkg/parser/ast",
         "//pkg/parser/mysql",
-        "//pkg/planner/core",
         "//pkg/planner/core/base",
+        "//pkg/planner/core/operator/logicalop",
         "//pkg/planner/property",
         "//pkg/sessionctx",
         "//pkg/sessionctx/variable",
diff --git a/pkg/executor/internal/testutil/window.go b/pkg/executor/internal/testutil/window.go
index d80d456064..cf3dfe85b8 100644
--- a/pkg/executor/internal/testutil/window.go
+++ b/pkg/executor/internal/testutil/window.go
@@ -21,7 +21,7 @@ import (
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
-	"github.com/pingcap/tidb/pkg/planner/core"
+	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/sessionctx/variable"
 	"github.com/pingcap/tidb/pkg/types"
@@ -31,7 +31,7 @@ import (
 // WindowTestCase has a fixed schema (col Double, partitionBy LongLong, rawData VarString(16), col LongLong).
 type WindowTestCase struct {
 	Ctx              sessionctx.Context
-	Frame            *core.WindowFrame
+	Frame            *logicalop.WindowFrame
 	WindowFunc       string
 	RawDataSmall     string
 	Columns          []*expression.Column
diff --git a/pkg/planner/core/casetest/partition/BUILD.bazel b/pkg/planner/core/casetest/partition/BUILD.bazel
index 70e2eb3e0b..b372c30815 100644
--- a/pkg/planner/core/casetest/partition/BUILD.bazel
+++ b/pkg/planner/core/casetest/partition/BUILD.bazel
@@ -10,7 +10,7 @@ go_test(
     ],
     data = glob(["testdata/**"]),
     flaky = True,
-    shard_count = 7,
+    shard_count = 9,
     deps = [
         "//pkg/config",
         "//pkg/planner/util/coretestsdk",
diff --git a/pkg/planner/core/casetest/partition/partition_pruner_test.go b/pkg/planner/core/casetest/partition/partition_pruner_test.go
index 23057c98fc..78d54174f0 100644
--- a/pkg/planner/core/casetest/partition/partition_pruner_test.go
+++ b/pkg/planner/core/casetest/partition/partition_pruner_test.go
@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"fmt"
 	"sort"
+	"strconv"
 	"strings"
 	"testing"
 
@@ -233,3 +234,294 @@ func TestPointGetIntHandleNotFirst(t *testing.T) {
 	tk.MustQuery("select * from t WHERE a BETWEEN 13 AND 13").Check(testkit.Rows("1 13 1"))
 	tk.MustQuery(`select * from t`).Check(testkit.Rows("1 13 1"))
 }
+
+type ExtractTestCase struct {
+	TimeUnit    string
+	ColumnTypes []string
+	PruneResult []string // cmpOps + BETWEEN pRange[1] [, pRange[2]]
+	NoFspResult string
+}
+
+// TODO: test LIST/HASH pruning?
+func TestRangeDatePruningExtract(t *testing.T) {
+	for _, colType := range []string{"DATE", "DATETIME", "DATETIME(1)", "DATETIME(6)"} {
+		extractTestCases := []ExtractTestCase{
+			{
+				"YEAR",
+				[]string{"DATE", "DATETIME"},
+				[]string{"p2", "p0,p1,p2", "p2,p3,pMax", "p0,p1,p2", "p2,p3,pMax", "p2,p3"},
+				"",
+			}, {
+				"QUARTER",
+				[]string{"DATE", "DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"YEAR_MONTH",
+				[]string{"DATE", "DATETIME"},
+				[]string{"p2", "p0,p1,p2", "p2,p3,pMax", "p0,p1,p2", "p2,p3,pMax", "p2,p3"},
+				"",
+			}, {
+
+				"MONTH",
+				[]string{"DATE", "DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"WEEK",
+				[]string{},
+				[]string{},
+				"",
+			}, {
+				"DAY",
+				[]string{"DATE", "DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"DAY_HOUR",
+				[]string{"DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"DAY_MINUTE",
+				[]string{"DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"DAY_SECOND",
+				[]string{"DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"DAY_MICROSECOND",
+				[]string{"DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				// Would also be affected by FSP truncation, but since the partition definitions
+				// in this test are increasing for each partition, this will not be noticed
+				"",
+			}, {
+				"HOUR",
+				[]string{"DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"HOUR_MINUTE",
+				[]string{"DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"HOUR_SECOND",
+				[]string{"DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"HOUR_MICROSECOND",
+				[]string{"DATETIME"},
+				// If no fsp is given, the partitioning expression still records
+				// the fsp, but evaluation will truncate it, so that is why
+				// the pruning will give p1, which is actually correct!
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"p1",
+			}, {
+				"MINUTE",
+				[]string{"DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"MINUTE_SECOND",
+				[]string{"DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"MINUTE_MICROSECOND",
+				[]string{"DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"p1",
+			}, {
+				"SECOND",
+				[]string{"DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"SECOND_MICROSECOND",
+				[]string{"DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"p1",
+			}, {
+				"MICROSECOND",
+				[]string{"DATETIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"p1",
+			},
+		}
+		runExtractTestCases(t, colType, extractTestCases)
+	}
+}
+
+func runExtractTestCases(t *testing.T, colType string, extractTestCases []ExtractTestCase) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+
+	// Loop over different datatypes, DATE, DATETIME(fsp), TIMESTAMP(fsp)
+	pRanges := []string{
+		"1990-01-01 00:00:00.000000",
+		"1991-04-02 01:01:01.100000",
+		"1992-08-03 02:02:02.200000",
+		"1993-12-31 23:59:59.999999",
+	}
+	cmpOps := []string{"=", "<", ">", "<=", ">="}
+	for _, tc := range extractTestCases {
+		found := false
+		hasFsp := strings.HasSuffix(colType, ")")
+		for _, cType := range tc.ColumnTypes {
+			end := strings.TrimPrefix(colType, cType)
+
+			if end == "" || end[:1] == "(" {
+				found = true
+			}
+		}
+		pRangesStrings := make([]string, 0, len(pRanges))
+		partDefs := ""
+		for i, pString := range pRanges {
+			r := tk.MustQuery(`SELECT EXTRACT(` + tc.TimeUnit + ` FROM '` + pString + `')`)
+			pRangesStrings = append(pRangesStrings, r.Rows()[0][0].(string))
+			if i > 0 {
+				partDefs += ", "
+			}
+			partDefs += "PARTITION p" +
+				strconv.Itoa(i) +
+				" VALUES LESS THAN (" +
+				pRangesStrings[i] + ")"
+		}
+		tk.MustExec(`drop table if exists t`)
+		createSQL := `create table t (d ` + colType + `, f varchar(255)) partition by range (EXTRACT(` + tc.TimeUnit + ` FROM d)) (` + partDefs + `, partition pMax values less than (maxvalue))`
+		if !found {
+			tk.MustContainErrMsg(createSQL, `[ddl:1486]Constant, random or timezone-dependent expressions in (sub)partitioning function are not allowed`)
+			continue
+		}
+		tk.MustExec(createSQL)
+		for i, op := range cmpOps {
+			res := tk.MustQuery(`explain select * from t where d ` + op + ` '` + pRanges[1] + `'`)
+			parts := strings.TrimPrefix(res.Rows()[0][3].(string), "partition:")
+			require.Greater(t, len(tc.PruneResult), i, "PruneResults does not include enough values, colType %s, EXTRACT %s, op %s", colType, tc.TimeUnit, op)
+			expects := tc.PruneResult[i]
+			if i == 0 && !hasFsp && tc.NoFspResult != "" {
+				expects = tc.NoFspResult
+			}
+			require.Equal(t, expects, parts, "colType %s, EXTRACT %s, op %s", colType, tc.TimeUnit, op)
+		}
+		res := tk.MustQuery(`explain select * from t where d between '` + pRanges[1] + `' and '` + pRanges[2] + `'`)
+		parts := strings.TrimPrefix(res.Rows()[0][3].(string), "partition:")
+		require.Equal(t, tc.PruneResult[len(cmpOps)], parts, "colType %s, EXTRACT %s, BETWEEN", colType, tc.TimeUnit)
+	}
+}
+
+func TestRangeTimePruningExtract(t *testing.T) {
+	for _, colType := range []string{"TIME", "TIME(1)", "TIME(6)", "TIMESTAMP", "TIMESTAMP(1)", "TIMESTAMP(6)"} {
+		extractTestCases := []ExtractTestCase{
+			{
+				"YEAR",
+				[]string{"DATE", "DATETIME"},
+				[]string{},
+				"",
+			}, {
+				"QUARTER",
+				[]string{"DATE", "DATETIME"},
+				[]string{},
+				"",
+			}, {
+				"YEAR_MONTH",
+				[]string{"DATE", "DATETIME"},
+				[]string{},
+				"",
+			}, {
+				"MONTH",
+				[]string{"DATE", "DATETIME"},
+				[]string{},
+				"",
+			}, {
+				"WEEK",
+				[]string{"DATE", "DATETIME"},
+				[]string{},
+				"",
+			}, {
+				"DAY",
+				[]string{"DATE", "DATETIME"},
+				[]string{},
+				"",
+			}, {
+				"DAY_HOUR",
+				[]string{"DATETIME"},
+				[]string{},
+				"",
+			}, {
+				"DAY_MINUTE",
+				[]string{"DATETIME"},
+				[]string{},
+				"",
+			}, {
+				"DAY_SECOND",
+				[]string{"DATETIME"},
+				[]string{},
+				"",
+			}, {
+				"DAY_MICROSECOND",
+				[]string{"DATETIME"},
+				[]string{},
+				"",
+			}, {
+				"HOUR",
+				[]string{"TIME"},
+				[]string{"p2", "p0,p1,p2", "p2,p3,pMax", "p0,p1,p2", "p2,p3,pMax", "p2,p3"},
+				"",
+			}, {
+				"HOUR_MINUTE",
+				[]string{"TIME"},
+				[]string{"p2", "p0,p1,p2", "p2,p3,pMax", "p0,p1,p2", "p2,p3,pMax", "p2,p3"},
+				"",
+			}, {
+				"HOUR_SECOND",
+				[]string{"TIME"},
+				[]string{"p2", "p0,p1,p2", "p2,p3,pMax", "p0,p1,p2", "p2,p3,pMax", "p2,p3"},
+				"",
+			}, {
+				"HOUR_MICROSECOND",
+				[]string{"TIME"},
+				[]string{"p2", "p0,p1,p2", "p2,p3,pMax", "p0,p1,p2", "p2,p3,pMax", "p2,p3"},
+				"",
+			}, {
+				"MINUTE",
+				[]string{"TIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"MINUTE_SECOND",
+				[]string{"TIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"MINUTE_MICROSECOND",
+				[]string{"TIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"SECOND",
+				[]string{"TIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"SECOND_MICROSECOND",
+				[]string{"TIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			}, {
+				"MICROSECOND",
+				[]string{"TIME"},
+				[]string{"p2", "all", "all", "all", "all", "all"},
+				"",
+			},
+		}
+		runExtractTestCases(t, colType, extractTestCases)
+	}
+}
diff --git a/pkg/testkit/testkit.go b/pkg/testkit/testkit.go
index a7dfd85443..fbfd9d8855 100644
--- a/pkg/testkit/testkit.go
+++ b/pkg/testkit/testkit.go
@@ -485,7 +485,7 @@ func (tk *TestKit) MustGetDBError(sql string, dberr *terror.Error) {
 // MustContainErrMsg executes a sql statement and assert its error message containing errStr.
 func (tk *TestKit) MustContainErrMsg(sql string, errStr any) {
 	err := tk.ExecToErr(sql)
-	tk.require.Error(err)
+	tk.require.Error(err, "sql: %s", sql)
 	tk.require.Contains(err.Error(), errStr)
 }