Repository: matrixorigin/matrixone
Pull request #9798 (2023-06-01 10:01:24)

CI: not store log in s3 and upload to summary
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [x] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #

## What this PR does / why we need it:
1. not store log in s3
2. upload logs to summary

matrix-meow: @guguducken Thanks for your contributions!

Title: CI: not store log in s3 and upload to summary

The title of the pull request is clear and concise. It indicates that the changes made in this pull request are related to the CI process and that logs will no longer be stored in s3, but will be uploaded to a summary instead.

Body:

The body of the pull request contains a checklist of different types of pull requests, and the author has checked the "Test and CI" option. The body also includes a section for the issue(s) that this pull request fixes, but it is left blank. The author has provided a brief description of what this pull request does and why it is needed. The author has also included the changes made in the pull request.

Changes:

The changes made in the pull request include removing the code that stores logs in s3 and adding code to upload logs to a summary. The author has also commented out a section of code that moves results to s3 if the test fails.

Issues:

1. The author has left the "Which issue(s) this PR fixes" section blank. It is important to include the issue number(s) that this pull request addresses to help with tracking and organization.

Suggestions:

1. The author should fill in the "Which issue(s) this PR fixes" section with the relevant issue number(s).
2. The author should provide more details in the body of the pull request about why the changes were made and how they will improve the CI process.
3. The author should consider adding comments to the code changes to explain what each change does and why it was made.
4. The author should consider adding tests to ensure that the changes made in the pull request do not introduce new bugs or issues.
5. The author should consider revisiting the section of code that was commented out and either remove it or provide an explanation for why it was commented out.

Optimizations:

1. The author could consider adding more detailed documentation to the code to help other developers understand how the CI process works and how to make changes in the future.
2. The author could consider adding more error handling to the code to ensure that any issues that arise during the CI process are handled gracefully and do not cause the process to fail.

Patch:
diff --git a/.github/workflows/mo_checkin_regression.yaml b/.github/workflows/mo_checkin_regression.yaml
index 4230e0ead..89fe4f9be 100644
--- a/.github/workflows/mo_checkin_regression.yaml
+++ b/.github/workflows/mo_checkin_regression.yaml
@@ -532,12 +532,16 @@ jobs:
           grep TPCC config.yml
           python Processor.py ${{ github.run_id }} ${{ needs.docker_image_build.outputs.version }} TPCC
           python Checker.py TPCC
+      
+      - uses: actions/upload-artifact@v3
+        if: ${{ failure() || cancelled()}}
+        continue-on-error: true
+        with:
+          name: mo-checkin-regression-${{ github.event.pull_request.number }}-${{ github.run_id }}
+          path: |
+             ${{ github.workspace }}/reports
+          retention-days: 7
 
-      - name: Test Plugin Result export
-        if: ${{ failure() || cancelled() }}
-        run: | 
-          aws s3 rm s3://mo-checkin-regression/result/${{ github.event.pull_request.number }}/reports --recursive
-          aws s3 cp $GITHUB_WORKSPACE/reports s3://mo-checkin-regression/result/${{ github.event.pull_request.number }}/reports --recursive
       - name : ISSUE Notice
         if: ${{ failure() || cancelled() }}
         uses: chf007/action-wechat-work@master
@@ -608,7 +612,7 @@ jobs:
 
 
           # move result to s3://mo-checkin-regression/result/ only test fail
-          if [ '${{ needs.test.result }}'x == 'failure'x ]; then aws s3 rm s3://mo-checkin-regression/result/${{ github.event.pull_request.number }}/etl --recursive > /dev/null 2>&1; aws s3 mv s3://mo-checkin-regression/${{ github.event.pull_request.number }}/etl s3://mo-checkin-regression/result/${{ github.event.pull_request.number }}/etl --recursive > /dev/null 2>&1; fi
+          #if [ '${{ needs.test.result }}'x == 'failure'x ]; then aws s3 rm s3://mo-checkin-regression/result/${{ github.event.pull_request.number }}/etl --recursive > /dev/null 2>&1; aws s3 mv s3://mo-checkin-regression/${{ github.event.pull_request.number }}/etl s3://mo-checkin-regression/result/${{ github.event.pull_request.number }}/etl --recursive > /dev/null 2>&1; fi
           kubectl get bucket -n mo-checkin-regression-${{ github.event.pull_request.number }} | grep bucket | awk '{system("kubectl delete bucket/"$1" -n mo-checkin-regression-${{ github.event.pull_request.number }}")}'
 
           kubectl delete namespace mo-checkin-regression-${{ github.event.pull_request.number }}
diff --git a/pkg/frontend/routine_manager.go b/pkg/frontend/routine_manager.go
index 76ec57edb..47e002b9e 100644
--- a/pkg/frontend/routine_manager.go
+++ b/pkg/frontend/routine_manager.go
@@ -67,7 +67,7 @@ func NewKillRecord(killtime time.Time, version uint64) KillRecord {
 }
 
 func (ar *AccountRoutineManager) recordRountine(tenantID int64, rt *Routine, version uint64) {
-	if tenantID == sysAccountID || rt != nil {
+	if tenantID == sysAccountID || rt == nil {
 		return
 	}
 
@@ -80,7 +80,7 @@ func (ar *AccountRoutineManager) recordRountine(tenantID int64, rt *Routine, ver
 }
 
 func (ar *AccountRoutineManager) deleteRoutine(tenantID int64, rt *Routine) {
-	if tenantID == sysAccountID || rt != nil {
+	if tenantID == sysAccountID || rt == nil {
 		return
 	}
 
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index fa225b445..2654ca64a 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -667,6 +667,34 @@ func (l *store) queryLog(ctx context.Context, shardID uint64,
 	}
 }
 
+func (l *store) tickerForTaskSchedule(ctx context.Context, duration time.Duration) {
+	ticker := time.NewTicker(duration)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			state, _ := l.getCheckerStateFromLeader()
+			if state != nil && state.State == pb.HAKeeperRunning {
+				l.taskSchedule(state)
+			}
+
+		case <-ctx.Done():
+			return
+		}
+
+		// l.taskSchedule could be blocking a long time, this extra select
+		// can give a chance immediately to check the ctx status when it resumes.
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			// nothing to do
+		}
+	}
+
+}
+
 func (l *store) ticker(ctx context.Context) {
 	if l.cfg.HAKeeperTickInterval.Duration == 0 {
 		panic("invalid HAKeeperTickInterval")
@@ -685,6 +713,12 @@ func (l *store) ticker(ctx context.Context) {
 	haTicker := time.NewTicker(l.cfg.HAKeeperCheckInterval.Duration)
 	defer haTicker.Stop()
 
+	// moving task schedule from the ticker normal routine to a
+	// separate goroutine can avoid the hakeeper's health check and tick update
+	// operations being blocked by task schedule, or the tick will be skipped and
+	// can not correctly estimate the time passing.
+	go l.tickerForTaskSchedule(ctx, l.cfg.HAKeeperCheckInterval.Duration)
+
 	for {
 		select {
 		case <-ticker.C:
diff --git a/pkg/logservice/store_hakeeper_check.go b/pkg/logservice/store_hakeeper_check.go
index 3a8230e25..006c8d7f2 100644
--- a/pkg/logservice/store_hakeeper_check.go
+++ b/pkg/logservice/store_hakeeper_check.go
@@ -111,23 +111,32 @@ func (l *store) updateIDAlloc(count uint64) error {
 	return nil
 }
 
-func (l *store) hakeeperCheck() {
+func (l *store) getCheckerStateFromLeader() (*pb.CheckerState, uint64) {
 	isLeader, term, err := l.isLeaderHAKeeper()
 	if err != nil {
 		l.runtime.Logger().Error("failed to get HAKeeper Leader ID", zap.Error(err))
-		return
+		return nil, term
 	}
 
 	if !isLeader {
 		l.taskScheduler.StopScheduleCronTask()
-		return
+		return nil, term
 	}
 	state, err := l.getCheckerState()
 	if err != nil {
 		// TODO: check whether this is temp error
 		l.runtime.Logger().Error("failed to get checker state", zap.Error(err))
+		return nil, term
+	}
+	return state, term
+}
+
+func (l *store) hakeeperCheck() {
+	state, term := l.getCheckerStateFromLeader()
+	if state == nil {
 		return
 	}
+
 	switch state.State {
 	case pb.HAKeeperCreated:
 		l.runtime.Logger().Warn("waiting for initial cluster info to be set, check skipped")
@@ -140,7 +149,6 @@ func (l *store) hakeeperCheck() {
 		l.handleBootstrapFailure()
 	case pb.HAKeeperRunning:
 		l.healthCheck(term, state)
-		l.taskSchedule(state)
 	default:
 		panic("unknown HAKeeper state")
 	}
diff --git a/pkg/sql/colexec/evalExpression.go b/pkg/sql/colexec/evalExpression.go
index 9d4d0d8aa..a12b9030c 100644
--- a/pkg/sql/colexec/evalExpression.go
+++ b/pkg/sql/colexec/evalExpression.go
@@ -78,7 +78,7 @@ type ExpressionExecutor interface {
 	// it will be called after query has done.
 	Free()
 
-	ifResultMemoryReuse() bool
+	IfResultMemoryReuse() bool
 }
 
 func NewExpressionExecutorsFromPlanExpressions(proc *process.Process, planExprs []*plan.Expr) (executors []ExpressionExecutor, err error) {
@@ -334,7 +334,7 @@ func (expr *FunctionExpressionExecutor) SetParameter(index int, executor Express
 	expr.parameterExecutor[index] = executor
 }
 
-func (expr *FunctionExpressionExecutor) ifResultMemoryReuse() bool {
+func (expr *FunctionExpressionExecutor) IfResultMemoryReuse() bool {
 	return true
 }
 
@@ -366,7 +366,7 @@ func (expr *ColumnExpressionExecutor) Free() {
 	// Nothing should do.
 }
 
-func (expr *ColumnExpressionExecutor) ifResultMemoryReuse() bool {
+func (expr *ColumnExpressionExecutor) IfResultMemoryReuse() bool {
 	return false
 }
 
@@ -393,7 +393,7 @@ func (expr *FixedVectorExpressionExecutor) Free() {
 	expr.resultVector = nil
 }
 
-func (expr *FixedVectorExpressionExecutor) ifResultMemoryReuse() bool {
+func (expr *FixedVectorExpressionExecutor) IfResultMemoryReuse() bool {
 	return true
 }
 
@@ -577,7 +577,7 @@ func FixProjectionResult(proc *process.Process, executors []ExpressionExecutor,
 			if oldVec.GetType().Oid == types.T_any {
 				newVec = vector.NewConstNull(types.T_any.ToType(), oldVec.Length(), proc.Mp())
 			} else {
-				if executors[i].ifResultMemoryReuse() {
+				if executors[i].IfResultMemoryReuse() {
 					if e, ok := executors[i].(*FunctionExpressionExecutor); ok {
 						// if projection, we can get the result directly
 						newVec = e.resultVector.GetResultVector()
@@ -631,7 +631,7 @@ func FixProjectionResult(proc *process.Process, executors []ExpressionExecutor,
 // I will remove this function later.
 // do not use this function.
 func SafeGetResult(proc *process.Process, vec *vector.Vector, executor ExpressionExecutor) (*vector.Vector, error) {
-	if executor.ifResultMemoryReuse() {
+	if executor.IfResultMemoryReuse() {
 		if e, ok := executor.(*FunctionExpressionExecutor); ok {
 			nv := e.resultVector.GetResultVector()
 			e.resultVector.SetResultVector(nil)
diff --git a/pkg/sql/colexec/window/window.go b/pkg/sql/colexec/window/window.go
index 2618c0a73..7a09d2f71 100644
--- a/pkg/sql/colexec/window/window.go
+++ b/pkg/sql/colexec/window/window.go
@@ -474,6 +474,14 @@ func (ctr *container) processOrder(idx int, ap *Argument, bat *batch.Batch, proc
 	if err := bat.Shuffle(sels, proc.Mp()); err != nil {
 		panic(err)
 	}
+
+	// shuffle agg vector
+	if ctr.aggVecs[idx].vec != nil && ctr.aggVecs[idx].executor.IfResultMemoryReuse() {
+		if err := ctr.aggVecs[idx].vec.Shuffle(sels, proc.Mp()); err != nil {
+			panic(err)
+		}
+	}
+
 	return false, nil
 }
 
diff --git a/pkg/vm/engine/disttae/logtailPushModel.go b/pkg/vm/engine/disttae/logtailPushModel.go
index 178e8be96..3cbee52fe 100644
--- a/pkg/vm/engine/disttae/logtailPushModel.go
+++ b/pkg/vm/engine/disttae/logtailPushModel.go
@@ -513,8 +513,10 @@ func (r *syncLogTailTimestamp) getTimestamp() timestamp.Timestamp {
 
 func (r *syncLogTailTimestamp) updateTimestamp(index int, newTimestamp timestamp.Timestamp) {
 	r.tList[index].Store(&newTimestamp)
-	ts := r.getTimestamp()
-	r.timestampWaiter.NotifyLatestCommitTS(ts)
+	if r.ready.Load() {
+		ts := r.getTimestamp()
+		r.timestampWaiter.NotifyLatestCommitTS(ts)
+	}
 }
 
 func (r *syncLogTailTimestamp) greatEq(txnTime timestamp.Timestamp) bool {
diff --git a/pkg/vm/engine/tae/db/replay.go b/pkg/vm/engine/tae/db/replay.go
index 45820bfa0..622064e14 100644
--- a/pkg/vm/engine/tae/db/replay.go
+++ b/pkg/vm/engine/tae/db/replay.go
@@ -42,6 +42,8 @@ type Replayer struct {
 	wg            sync.WaitGroup
 	applyDuration time.Duration
 	txnCmdChan    chan *txnbase.TxnCmd
+	readCount     int
+	applyCount    int
 }
 
 func newReplayer(dataFactory *tables.DataFactory, db *DB, ckpedTS types.TS) *Replayer {
@@ -89,7 +91,9 @@ func (replayer *Replayer) Replay() {
 	replayer.wg.Wait()
 	logutil.Info("open-tae", common.OperationField("replay"),
 		common.OperandField("wal"),
-		common.AnyField("apply logentries cost", replayer.applyDuration))
+		common.AnyField("apply logentries cost", replayer.applyDuration),
+		common.AnyField("read count", replayer.readCount),
+		common.AnyField("apply count", replayer.applyCount))
 }
 
 func (replayer *Replayer) OnReplayEntry(group uint32, lsn uint64, payload []byte, typ uint16, info any) {
@@ -133,10 +137,12 @@ func (replayer *Replayer) OnTimeStamp(ts types.TS) {
 
 func (replayer *Replayer) OnReplayTxn(cmd txnif.TxnCmd, lsn uint64) {
 	var err error
+	replayer.readCount++
 	txnCmd := cmd.(*txnbase.TxnCmd)
 	if txnCmd.PrepareTS.LessEq(replayer.maxTs) {
 		return
 	}
+	replayer.applyCount++
 	txn := txnimpl.MakeReplayTxn(replayer.db.TxnMgr, txnCmd.TxnCtx, lsn,
 		txnCmd, replayer, replayer.db.Catalog, replayer.DataFactory, replayer.db.Wal)
 	if err = replayer.db.TxnMgr.OnReplayTxn(txn); err != nil {
diff --git a/pkg/vm/engine/tae/logstore/driver/logservicedriver/driver.go b/pkg/vm/engine/tae/logstore/driver/logservicedriver/driver.go
index c9cd61c5c..8a7865b2f 100644
--- a/pkg/vm/engine/tae/logstore/driver/logservicedriver/driver.go
+++ b/pkg/vm/engine/tae/logstore/driver/logservicedriver/driver.go
@@ -123,7 +123,11 @@ func (d *LogServiceDriver) Replay(h driver.ApplyHandle) error {
 		common.OperandField("wal"),
 		common.AnyField("backend", "logservice"),
 		common.AnyField("apply cost", r.applyDuration),
-		common.AnyField("read cost", d.readDuration))
+		common.AnyField("read cost", d.readDuration),
+		common.AnyField("read count", r.readCount),
+		common.AnyField("internal count", r.internalCount),
+		common.AnyField("apply count", r.applyCount),
+	)
 
 	return nil
 }
diff --git a/pkg/vm/engine/tae/logstore/driver/logservicedriver/replay.go b/pkg/vm/engine/tae/logstore/driver/logservicedriver/replay.go
index 6ad0a88ec..205318cd6 100644
--- a/pkg/vm/engine/tae/logstore/driver/logservicedriver/replay.go
+++ b/pkg/vm/engine/tae/logstore/driver/logservicedriver/replay.go
@@ -50,6 +50,9 @@ type replayer struct {
 	recordChan chan *entry.Entry
 
 	applyDuration time.Duration
+	readCount     int
+	internalCount int
+	applyCount    int
 
 	wg sync.WaitGroup
 }
@@ -97,7 +100,9 @@ func (r *replayer) replay() {
 
 func (r *replayer) readRecords() (readEnd bool) {
 	nextLsn, safeLsn := r.d.readFromLogServiceInReplay(r.nextToReadLsn, r.readMaxSize, func(lsn uint64, record *recordEntry) {
+		r.readCount++
 		if record.meta.metaType == TReplay {
+			r.internalCount++
 			cmd := NewEmptyReplayCmd()
 			cmd.Unmarshal(record.payload)
 			r.removeEntries(cmd.skipLsns)
@@ -167,6 +172,7 @@ func (r *replayer) replayLogserviceEntry(lsn uint64, safe bool) error {
 	if err != nil {
 		panic(err)
 	}
+	r.applyCount++
 	intervals := record.replay(r)
 	r.d.onReplayRecordEntry(logserviceLsn, intervals)
 	r.onReplayDriverLsn(intervals.GetMax())
diff --git a/pkg/vm/engine/tae/logtail/service/session.go b/pkg/vm/engine/tae/logtail/service/session.go
index 5f84dcaea..3129454be 100644
--- a/pkg/vm/engine/tae/logtail/service/session.go
+++ b/pkg/vm/engine/tae/logtail/service/session.go
@@ -330,7 +330,7 @@ func (ss *Session) FilterLogtail(tails ...wrapLogtail) []logtail.TableLogtail {
 		if state, ok := ss.tables[t.id]; ok && state == TableSubscribed {
 			qualified = append(qualified, t.tail)
 		} else {
-			ss.logger.Info("table not subscribed, filter out",
+			ss.logger.Debug("table not subscribed, filter out",
 				zap.String("id", string(t.id)),
 				zap.String("checkpoint", t.tail.CkpLocation),
 			)


Test patch:
diff --git a/pkg/logservice/store_hakeeper_check_test.go b/pkg/logservice/store_hakeeper_check_test.go
index 30f9feffa..227727860 100644
--- a/pkg/logservice/store_hakeeper_check_test.go
+++ b/pkg/logservice/store_hakeeper_check_test.go
@@ -494,6 +494,37 @@ func TestHAKeeperCanBootstrapAndRepairShards(t *testing.T) {
 	runHAKeeperClusterTest(t, fn)
 }
 
+func TestGetCheckerStateFromLeader(t *testing.T) {
+	fn := func(t *testing.T, store *store) {
+		ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Second*10))
+		defer cancel()
+
+		for {
+			select {
+			case <-ctx.Done():
+				t.Error("test deadline reached")
+				return
+
+			default:
+				isLeader, termA, err := store.isLeaderHAKeeper()
+				state, termB := store.getCheckerStateFromLeader()
+				require.NoError(t, err)
+				assert.Equal(t, termB, termA)
+
+				if !isLeader {
+					assert.Equal(t, (*pb.CheckerState)(nil), state)
+				} else {
+					assert.NotEqual(t, (*pb.CheckerState)(nil), state)
+					return
+				}
+				time.Sleep(time.Second)
+			}
+		}
+	}
+
+	runHAKeeperStoreTest(t, false, fn)
+}
+
 func TestGetCheckerState(t *testing.T) {
 	fn := func(t *testing.T, store *store) {
 		state, err := store.getCheckerState()
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index 49d44ef55..051c436da 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -17,6 +17,7 @@ package logservice
 import (
 	"context"
 	"fmt"
+	"github.com/matrixorigin/matrixone/pkg/pb/task"
 	"math"
 	"sync/atomic"
 	"testing"
@@ -298,6 +299,97 @@ func TestQueryLog(t *testing.T) {
 	runStoreTest(t, fn)
 }
 
+func proceedHAKeeperToRunning(t *testing.T, store *store) {
+	state, err := store.getCheckerState()
+	assert.NoError(t, err)
+	assert.Equal(t, pb.HAKeeperCreated, state.State)
+
+	err = store.setInitialClusterInfo(1, 1, 1)
+	assert.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	hb := store.getHeartbeatMessage()
+	_, err = store.addLogStoreHeartbeat(ctx, hb)
+	assert.NoError(t, err)
+
+	state, err = store.getCheckerState()
+	assert.NoError(t, err)
+	assert.Equal(t, pb.HAKeeperBootstrapping, state.State)
+
+	_, term, err := store.isLeaderHAKeeper()
+	assert.NoError(t, err)
+
+	store.bootstrap(term, state)
+	state, err = store.getCheckerState()
+
+	assert.NoError(t, err)
+	assert.Equal(t, pb.HAKeeperBootstrapCommandsReceived, state.State)
+
+	cmd, err := store.getCommandBatch(ctx, store.id())
+	require.NoError(t, err)
+	require.Equal(t, 1, len(cmd.Commands))
+	assert.True(t, cmd.Commands[0].Bootstrapping)
+
+	// handle startReplica to make sure logHeartbeat msg contain shards info,
+	// which used in store.checkBootstrap to determine if all log shards ready
+	service := &Service{store: store}
+	service.handleStartReplica(cmd.Commands[0])
+
+	for state.State != pb.HAKeeperRunning && store.bootstrapCheckCycles > 0 {
+		func() {
+			ctx, cancel = context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
+
+			_, err = store.addLogStoreHeartbeat(ctx, store.getHeartbeatMessage())
+			assert.NoError(t, err)
+
+			store.checkBootstrap(state)
+			state, err = store.getCheckerState()
+			assert.NoError(t, err)
+
+			time.Sleep(time.Millisecond * 100)
+		}()
+	}
+
+	assert.Equal(t, pb.HAKeeperRunning, state.State)
+}
+
+// test if the tickerForTaskSchedule can push forward these routine
+func TestTickerForTaskSchedule(t *testing.T) {
+	fn := func(t *testing.T, store *store, taskService taskservice.TaskService) {
+		// making hakeeper state proceeds to running before test task schedule
+		tickerCxt, tickerCancel := context.WithCancel(context.Background())
+		defer tickerCancel()
+
+		//do task schedule background
+		go store.tickerForTaskSchedule(tickerCxt, time.Millisecond*10)
+
+		proceedHAKeeperToRunning(t, store)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		err := taskService.Create(ctx, task.TaskMetadata{ID: "1234"})
+		assert.NoError(t, err)
+
+		cnUUID := uuid.New().String()
+		cmd := pb.CNStoreHeartbeat{UUID: cnUUID}
+		_, err = store.addCNStoreHeartbeat(ctx, cmd)
+		assert.NoError(t, err)
+
+		// waiting the background taskSchedule operation
+		// schedule task that we created to CN node
+		time.Sleep(time.Millisecond * 100)
+
+		tasks, err := taskService.QueryTask(ctx, taskservice.WithTaskRunnerCond(taskservice.EQ, cnUUID))
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(tasks))
+	}
+
+	runHakeeperTaskServiceTest(t, fn)
+}
+
 func TestHAKeeperTick(t *testing.T) {
 	fn := func(t *testing.T, store *store) {
 		peers := make(map[uint64]dragonboat.Target)
diff --git a/test/distributed/cases/window/window.result b/test/distributed/cases/window/window.result
index f1ff606fa..7bbdc60da 100644
--- a/test/distributed/cases/window/window.result
+++ b/test/distributed/cases/window/window.result
@@ -447,4 +447,40 @@ a    sum(b) over (partition by a order by a)
 1    12.12
 2    123.13
 3    456.66
-4    1111.34
\ No newline at end of file
+4    1111.34
+drop table if exists wf01;
+create table wf01(i int,j int);
+insert into wf01 values(1,1);
+insert into wf01 values(1,4);
+insert into wf01 values(1,2);
+insert into wf01 values(1,4);
+select * from wf01;
+i    j
+1    1
+1    4
+1    2
+1    4
+select i, j, sum(i+j) over (order by j desc rows unbounded preceding) foo from wf01;
+i    j    foo
+1    4    5
+1    4    10
+1    2    13
+1    1    15
+select i, j, sum(i+j) over (order by j desc rows between 2 preceding and 2 following) as foo from wf01;
+i    j    foo
+1    4    13
+1    4    15
+1    2    15
+1    1    10
+select i, j, sum(i+j) over (order by j desc rows unbounded preceding) foo from wf01 order by foo;
+i    j    foo
+1    4    5
+1    4    10
+1    2    13
+1    1    15
+select i, j, sum(i+j) over (order by j desc rows unbounded preceding) foo from wf01 order by foo desc;
+i    j    foo
+1    1    15
+1    2    13
+1    4    10
+1    4    5
\ No newline at end of file
diff --git a/test/distributed/cases/window/window.sql b/test/distributed/cases/window/window.sql
index 963e01967..0195168d0 100644
--- a/test/distributed/cases/window/window.sql
+++ b/test/distributed/cases/window/window.sql
@@ -69,4 +69,16 @@ select a, c, rank() over(order by a), row_number() over(order by a), dense_rank(
 drop table if exists t1;
 create table t1 (a int, b decimal(7, 2));
 insert into t1 values(1, 12.12), (2, 123.13), (3, 456.66), (4, 1111.34);
-select a, sum(b) over (partition by a order by a) from t1;
\ No newline at end of file
+select a, sum(b) over (partition by a order by a) from t1;
+
+drop table if exists wf01;
+create table wf01(i int,j int);
+insert into wf01 values(1,1);
+insert into wf01 values(1,4);
+insert into wf01 values(1,2);
+insert into wf01 values(1,4);
+select * from wf01;
+select i, j, sum(i+j) over (order by j desc rows unbounded preceding) foo from wf01;
+select i, j, sum(i+j) over (order by j desc rows between 2 preceding and 2 following) as foo from wf01;
+select i, j, sum(i+j) over (order by j desc rows unbounded preceding) foo from wf01 order by foo;
+select i, j, sum(i+j) over (order by j desc rows unbounded preceding) foo from wf01 order by foo desc;
\ No newline at end of file