core: vm: sqlvm: add opRange

Add `opRange` which supports row range evaluation, such as `LIMIT` and
`OFFSET`.

5 files changed, 318 insertions, 0 deletions

diff --git a/core/vm/sqlvm/runtime/instructions.go b/core/vm/sqlvm/runtime/instructions.go
index cc00f741e..353701f8c 100644
--- a/core/vm/sqlvm/runtime/instructions.go
+++ b/core/vm/sqlvm/runtime/instructions.go
@@ -1398,6 +1398,42 @@ func opCut(ctx *common.Context, ops, registers []*Operand, output uint) (err err
 }
 
 // in-place Op
+func opRange(ctx *common.Context, ops, registers []*Operand, output uint) (err error) {
+	if len(ops) != 2 {
+		err = se.ErrorCodeInvalidOperandNum
+		return
+	}
+	op, slice := ops[0], ops[1].Data[0]
+
+	offset, err := ast.DecimalToUint64(slice[0].Value)
+	if err != nil {
+		return
+	}
+
+	if offset < uint64(len(op.Data)) {
+		op.Data = op.Data[offset:]
+	} else {
+		op.Data = op.Data[:0]
+	}
+
+	if len(slice) > 1 && len(op.Data) > 0 {
+		var limit uint64
+
+		limit, err = ast.DecimalToUint64(slice[1].Value)
+		if err != nil {
+			return
+		}
+
+		if limit < uint64(len(op.Data)) {
+			op.Data = op.Data[:limit]
+		}
+	}
+
+	registers[output] = op
+	return
+}
+
+// in-place Op
 func opSort(ctx *common.Context, ops, registers []*Operand, output uint) (err error) {
 	if len(ops) != 2 {
 		err = se.ErrorCodeInvalidOperandNum
diff --git a/core/vm/sqlvm/runtime/instructions_op_test.go b/core/vm/sqlvm/runtime/instructions_op_test.go
index 0255e74cf..e0550006f 100644
--- a/core/vm/sqlvm/runtime/instructions_op_test.go
+++ b/core/vm/sqlvm/runtime/instructions_op_test.go
@@ -3337,3 +3337,163 @@ func (s *instructionSuite) TestOpSort() {
 
 	s.run(testcases, opSort)
 }
+
+func (s *instructionSuite) TestOpRange() {
+	testcases := []opTestcase{
+		{
+			"Range test limit 2 offset 1",
+			Instruction{
+				Op: RANGE,
+				Input: []*Operand{
+					makeOperand(
+						false,
+						[]ast.DataType{
+							ast.ComposeDataType(ast.DataTypeMajorInt, 0),
+						},
+						[]Tuple{
+							{&Raw{Value: decimal.NewFromFloat(1)}},
+							{&Raw{Value: decimal.NewFromFloat(2)}},
+							{&Raw{Value: decimal.NewFromFloat(3)}},
+							{&Raw{Value: decimal.NewFromFloat(4)}},
+							{&Raw{Value: decimal.NewFromFloat(5)}},
+							{&Raw{Value: decimal.NewFromFloat(6)}},
+							{&Raw{Value: decimal.NewFromFloat(7)}},
+							{&Raw{Value: decimal.NewFromFloat(8)}},
+						},
+					),
+					makeOperand(
+						true,
+						[]ast.DataType{
+							ast.ComposeDataType(ast.DataTypeMajorUint, 1), ast.ComposeDataType(ast.DataTypeMajorUint, 1),
+						},
+						[]Tuple{
+							{&Raw{Value: decimal.NewFromFloat(1)}, &Raw{Value: decimal.NewFromFloat(2)}},
+						},
+					),
+				},
+				Output: 0,
+			},
+			makeOperand(
+				false,
+				[]ast.DataType{
+					ast.ComposeDataType(ast.DataTypeMajorInt, 0),
+				},
+				[]Tuple{
+					{&Raw{Value: decimal.NewFromFloat(2)}},
+					{&Raw{Value: decimal.NewFromFloat(3)}},
+				},
+			),
+			nil,
+		},
+		{
+			"Range test limit 0 offset 1",
+			Instruction{
+				Op: RANGE,
+				Input: []*Operand{
+					makeOperand(
+						false,
+						[]ast.DataType{
+							ast.ComposeDataType(ast.DataTypeMajorInt, 0),
+						},
+						[]Tuple{
+							{&Raw{Value: decimal.NewFromFloat(1)}},
+							{&Raw{Value: decimal.NewFromFloat(2)}},
+						},
+					),
+					makeOperand(
+						true,
+						[]ast.DataType{
+							ast.ComposeDataType(ast.DataTypeMajorUint, 1), ast.ComposeDataType(ast.DataTypeMajorUint, 1),
+						},
+						[]Tuple{
+							{&Raw{Value: decimal.NewFromFloat(1)}, &Raw{Value: decimal.NewFromFloat(0)}},
+						},
+					),
+				},
+				Output: 0,
+			},
+			makeOperand(
+				false,
+				[]ast.DataType{
+					ast.ComposeDataType(ast.DataTypeMajorInt, 0),
+				},
+				[]Tuple{},
+			),
+			nil,
+		},
+		{
+			"Range test offset 20",
+			Instruction{
+				Op: RANGE,
+				Input: []*Operand{
+					makeOperand(
+						false,
+						[]ast.DataType{
+							ast.ComposeDataType(ast.DataTypeMajorInt, 0),
+						},
+						[]Tuple{
+							{&Raw{Value: decimal.NewFromFloat(1)}},
+							{&Raw{Value: decimal.NewFromFloat(2)}},
+						},
+					),
+					makeOperand(
+						true,
+						[]ast.DataType{
+							ast.ComposeDataType(ast.DataTypeMajorUint, 1),
+						},
+						[]Tuple{
+							{&Raw{Value: decimal.NewFromFloat(20)}},
+						},
+					),
+				},
+				Output: 0,
+			},
+			makeOperand(
+				false,
+				[]ast.DataType{
+					ast.ComposeDataType(ast.DataTypeMajorInt, 0),
+				},
+				[]Tuple{},
+			),
+			nil,
+		},
+		{
+			"Range test limit 10 offset 20",
+			Instruction{
+				Op: RANGE,
+				Input: []*Operand{
+					makeOperand(
+						false,
+						[]ast.DataType{
+							ast.ComposeDataType(ast.DataTypeMajorInt, 0),
+						},
+						[]Tuple{
+							{&Raw{Value: decimal.NewFromFloat(1)}},
+							{&Raw{Value: decimal.NewFromFloat(2)}},
+						},
+					),
+					makeOperand(
+						true,
+						[]ast.DataType{
+							ast.ComposeDataType(ast.DataTypeMajorUint, 1), ast.ComposeDataType(ast.DataTypeMajorUint, 1),
+						},
+						[]Tuple{
+							{&Raw{Value: decimal.NewFromFloat(20)}, &Raw{Value: decimal.NewFromFloat(10)}},
+						},
+					),
+				},
+				Output: 0,
+			},
+			makeOperand(
+				false,
+				[]ast.DataType{
+					ast.ComposeDataType(ast.DataTypeMajorInt, 0),
+				},
+				[]Tuple{},
+			),
+			nil,
+		},
+	}
+
+	s.run(testcases, opRange)
+}
diff --git a/core/vm/sqlvm/runtime/instructions_tmpl_data.go b/core/vm/sqlvm/runtime/instructions_tmpl_data.go
index c4a7ac4a7..7edc2c3e6 100644
--- a/core/vm/sqlvm/runtime/instructions_tmpl_data.go
+++ b/core/vm/sqlvm/runtime/instructions_tmpl_data.go
@@ -2330,5 +2330,125 @@ var testData = &tmplData{
 			},
 		},
 		// -- end of SORT
+		{
+			TestName: "OpRange", OpFunc: "opRange",
+			Cases: []*tmplTestCase{
+				{
+					Name:  "Range test limit 2 offset 1",
+					Error: "nil", OpCode: "RANGE",
+					Inputs: []*tmplOp{
+						{
+							Im: false,
+							Metas: []*tmplOpMeta{
+								{Major: "Int", Minor: 0},
+							},
+							Data: []string{
+								`{V: 1}`, `{V: 2}`, `{V: 3}`, `{V: 4}`,
+								`{V: 5}`, `{V: 6}`, `{V: 7}`, `{V: 8}`,
+							},
+						},
+						{
+							Im: true,
+							Metas: []*tmplOpMeta{
+								{Major: "Uint", Minor: 7},
+								{Major: "Uint", Minor: 7},
+							},
+							Data: []string{"{V: 1, V: 2}"},
+						},
+					},
+					Output: &tmplOp{
+						Im: false,
+						Metas: []*tmplOpMeta{
+							{Major: "Int", Minor: 0},
+						},
+						Data: []string{`{V: 2}`, `{V: 3}`},
+					},
+				},
+				{
+					Name:  "Range test limit 0 offset 1",
+					Error: "nil", OpCode: "RANGE",
+					Inputs: []*tmplOp{
+						{
+							Im: false,
+							Metas: []*tmplOpMeta{
+								{Major: "Int", Minor: 0},
+							},
+							Data: []string{`{V: 1}`, `{V: 2}`},
+						},
+						{
+							Im: true,
+							Metas: []*tmplOpMeta{
+								{Major: "Uint", Minor: 7},
+								{Major: "Uint", Minor: 7},
+							},
+							Data: []string{"{V: 1, V: 0}"},
+						},
+					},
+					Output: &tmplOp{
+						Im: false,
+						Metas: []*tmplOpMeta{
+							{Major: "Int", Minor: 0},
+						},
+						Data: []string{},
+					},
+				},
+				{
+					Name:  "Range test offset 20",
+					Error: "nil", OpCode: "RANGE",
+					Inputs: []*tmplOp{
+						{
+							Im: false,
+							Metas: []*tmplOpMeta{
+								{Major: "Int", Minor: 0},
+							},
+							Data: []string{`{V: 1}`, `{V: 2}`},
+						},
+						{
+							Im: true,
+							Metas: []*tmplOpMeta{
+								{Major: "Uint", Minor: 7},
+							},
+							Data: []string{"{V: 20}"},
+						},
+					},
+					Output: &tmplOp{
+						Im: false,
+						Metas: []*tmplOpMeta{
+							{Major: "Int", Minor: 0},
+						},
+						Data: []string{},
+					},
+				},
+				{
+					Name:  "Range test limit 10 offset 20",
+					Error: "nil", OpCode: "RANGE",
+					Inputs: []*tmplOp{
+						{
+							Im: false,
+							Metas: []*tmplOpMeta{
+								{Major: "Int", Minor: 0},
+							},
+							Data: []string{`{V: 1}`, `{V: 2}`},
+						},
+						{
+							Im: true,
+							Metas: []*tmplOpMeta{
+								{Major: "Uint", Minor: 7},
+								{Major: "Uint", Minor: 7},
+							},
+							Data: []string{"{V: 20, V: 10}"},
+						},
+					},
+					Output: &tmplOp{
+						Im: false,
+						Metas: []*tmplOpMeta{
+							{Major: "Int", Minor: 0},
+						},
+						Data: []string{},
+					},
+				},
+			},
+		},
+		// -- end of RANGE
 	},
 }
diff --git a/core/vm/sqlvm/runtime/jumptable.go b/core/vm/sqlvm/runtime/jumptable.go
index 2408ee0f4..aab57225d 100644
--- a/core/vm/sqlvm/runtime/jumptable.go
+++ b/core/vm/sqlvm/runtime/jumptable.go
@@ -27,6 +27,7 @@ var jumpTable = [256]OpFunction{
 	FILTER: opFilter,
 	CAST:   opCast,
 	CUT:    opCut,
+	RANGE:  opRange,
 
 	// 0x60
 	LOAD: opLoad,
diff --git a/core/vm/sqlvm/runtime/opcodes.go b/core/vm/sqlvm/runtime/opcodes.go
index 07b30bdc3..8827f0531 100644
--- a/core/vm/sqlvm/runtime/opcodes.go
+++ b/core/vm/sqlvm/runtime/opcodes.go
@@ -46,6 +46,7 @@ const (
 	FILTER
 	CAST
 	CUT
+	RANGE
 )
 
 // 0x60 range - storage ops