From 082b6e661c595f9e2d5e12e9975873cbf8561875 Mon Sep 17 00:00:00 2001 From: pokeeffe-molecula Date: Fri, 7 Apr 2023 16:40:07 -0500 Subject: [PATCH] introduced vector(n) type; inserts work --- batch/batch.go | 53 ++++++++++++++++++++-------- dax/table.go | 4 ++- field.go | 29 +++++++++++++++- holder.go | 2 +- http_handler.go | 6 ++++ index.go | 24 +++++++++---- schema.go | 7 ++++ sql3/errors.go | 8 +++++ sql3/parser/ast.go | 18 +++++----- sql3/parser/ast_test.go | 2 +- sql3/parser/astdatatype.go | 53 ++++++++++++++++++++++++++-- sql3/parser/parser.go | 4 +-- sql3/parser/token.go | 2 -- sql3/planner/compilecreatemodel.go | 2 +- sql3/planner/compilecreatetable.go | 14 ++++++++ sql3/planner/expression.go | 55 +++++++++++++++++++++--------- sql3/planner/expression_it_test.go | 2 +- sql3/planner/expressionanalyzer.go | 17 ++++++--- sql3/planner/expressionpql.go | 4 +-- sql3/planner/expressiontypes.go | 15 ++++++++ sql3/planner/opbulkinsert.go | 4 +-- sql3/planner/opcopy.go | 4 +-- sql3/planner/opinsert.go | 12 +++++++ tstore/btree.go | 3 ++ tstore/btreetypes.go | 36 +++++++++++++++++++ wireprotocol/wireprimitives.go | 54 ++++++++++++++++++++++++++++- 26 files changed, 362 insertions(+), 72 deletions(-) diff --git a/batch/batch.go b/batch/batch.go index 211c190d7..57f208ad4 100644 --- a/batch/batch.go +++ b/batch/batch.go @@ -133,7 +133,7 @@ type Batch struct { // values holds the values for each record of an int field values map[string][]int64 - // values holds the values for each record of an varchar field + // values holds the values for each record of an t-store field tupleValues map[string][]interface{} // boolValues is a map[fieldName][idsIndex]bool, which holds the values for @@ -304,6 +304,8 @@ func NewBatch(importer featurebase.Importer, size int, tbl *dax.Table, fields [] boolValues[field.Name] = make(map[int]bool) case featurebase.FieldTypeVarchar: tupleValues[field.Name] = make([]interface{}, 0, size) + case featurebase.FieldTypeVector: + tupleValues[field.Name] = make([]interface{}, 0, size) default: return nil, errors.Errorf("field type '%s' is not currently supported through Batch", typ) } @@ -666,6 +668,9 @@ func (b *Batch) Add(rec Row) error { case featurebase.FieldTypeVarchar: b.tupleValues[field.Name] = append(b.tupleValues[field.Name], nil) + case featurebase.FieldTypeVector: + b.tupleValues[field.Name] = append(b.tupleValues[field.Name], nil) + default: // only append nil to rowIDs if this field already has // rowIDs. Otherwise, this could be a []string or @@ -684,6 +689,14 @@ func (b *Batch) Add(rec Row) error { case pql.Decimal: b.values[field.Name] = append(b.values[field.Name], val.ToInt64(field.Options.Scale)) + case []float64: + switch field.Options.Type { + case featurebase.FieldTypeVector: + b.tupleValues[field.Name] = append(b.tupleValues[field.Name], val) + default: + return errors.Errorf("Val %v Type %[1]T is not currently supported. Use string, uint64 (row id), or int64 (integer value)", val) + } + default: return errors.Errorf("Val %v Type %[1]T is not currently supported. Use string, uint64 (row id), or int64 (integer value)", val) } @@ -1820,31 +1833,41 @@ func (b *Batch) makeTupleStoreFragments(pvlfrags tuplefragments) (tuplefragments AliasName: "", Type: parser.NewDataTypeVarchar(field.Options.Length), }) + case featurebase.FieldTypeVector: + tupleSchema = append(tupleSchema, &types.PlannerColumn{ + ColumnName: fieldname, + RelationName: string(b.tbl.Name), + AliasName: "", + Type: parser.NewDataTypeVector(field.Options.Length), + }) default: continue } } - // for each of the varcharValues mapped, this is a column - for fieldname, varcharMap := range b.tupleValues { + // for each of the values mapped, this is a column + for fieldname, tstoreMap := range b.tupleValues { field := b.headerMap[fieldname] - if field.Options.Type != featurebase.FieldTypeVarchar { - continue - } - // for each of the row values for this column - for pos, varcharVal := range varcharMap { - recID := b.ids[pos] + switch field.Options.Type { + case featurebase.FieldTypeVarchar, featurebase.FieldTypeVector: + // for each of the row values for this column + for pos, tstoreVal := range tstoreMap { + recID := b.ids[pos] - shard := recID / shardWidth - tf := pvlfrags.GetOrCreate(shard, tupleSchema) + shard := recID / shardWidth + tf := pvlfrags.GetOrCreate(shard, tupleSchema) + + _, ok := tf.tupleData[recID] + if !ok { + tf.tupleData[recID] = make(map[string]interface{}) + } + tf.tupleData[recID][fieldname] = tstoreVal - _, ok := tf.tupleData[recID] - if !ok { - tf.tupleData[recID] = make(map[string]interface{}) } - tf.tupleData[recID][fieldname] = varcharVal + default: + continue } } diff --git a/dax/table.go b/dax/table.go index 8cbd5c08f..076ae9a55 100644 --- a/dax/table.go +++ b/dax/table.go @@ -84,6 +84,7 @@ const ( BaseTypeStringSetQ = "stringsetq" // keyed set timequantum BaseTypeTimestamp = "timestamp" // BaseTypeVarchar = "varchar" // + BaseTypeVector = "vector" // DefaultPartitionN = 256 @@ -686,7 +687,8 @@ func BaseTypeFromString(s string) (BaseType, error) { BaseTypeStringSet, BaseTypeStringSetQ, BaseTypeTimestamp, - BaseTypeVarchar: + BaseTypeVarchar, + BaseTypeVector: return BaseType(lowered), nil default: return "", errors.Errorf("invalid field type: %s", s) diff --git a/field.go b/field.go index 2a3222ce1..377148910 100644 --- a/field.go +++ b/field.go @@ -34,6 +34,7 @@ const ( DefaultCacheSize = 50000 DefaultVarcharLength = 50 + DefaultVectorLength = 128 bitsPerWord = 32 << (^uint(0) >> 63) // either 32 or 64 maxInt = 1<<(bitsPerWord-1) - 1 // either 1<<31 - 1 or 1<<63 - 1 @@ -50,6 +51,7 @@ const ( FieldTypeDecimal = "decimal" FieldTypeTimestamp = "timestamp" FieldTypeVarchar = "varchar" + FieldTypeVector = "vector" ) type protected struct { @@ -419,6 +421,20 @@ func OptFieldTypeVarchar(length int64) FieldOption { } } +// OptFieldTypeVector is a functional option on FieldOptions +// used to specify the field as being type `vector` and to +// provide any respective configuration values. +func OptFieldTypeVector(length int64) FieldOption { + return func(fo *FieldOptions) error { + if fo.Type != "" { + return errors.Errorf("field type is already set to: %s", fo.Type) + } + fo.Type = FieldTypeVector + fo.Length = length + return nil + } +} + // newField returns a new instance of field (without name validation). func newField(holder *Holder, path, index, name string, opts ...FieldOption) (*Field, error) { // Apply functional option. @@ -612,7 +628,7 @@ func (f *Field) Open() error { return errors.Wrap(err, "creating field dir") } - if strings.EqualFold(f.options.Type, FieldTypeVarchar) { + if strings.EqualFold(f.options.Type, FieldTypeVarchar) || strings.EqualFold(f.options.Type, FieldTypeVector) { f.holder.Logger.Debugf("opening b-tree for index/field: %s/%s", f.index, f.name) // Apply the field options loaded from etcd (or set via setOptions()). @@ -927,6 +943,9 @@ func (f *Field) applyOptions(opt FieldOptions) error { case FieldTypeVarchar: f.options.Type = FieldTypeVarchar f.options.Length = opt.Length + case FieldTypeVector: + f.options.Type = FieldTypeVector + f.options.Length = opt.Length default: return errors.New("invalid field type") } @@ -2413,6 +2432,14 @@ func (o *FieldOptions) MarshalJSON() ([]byte, error) { o.Type, o.Length, }) + case FieldTypeVector: + return json.Marshal(struct { + Type string `json:"type"` + Length int64 `json:"length"` + }{ + o.Type, + o.Length, + }) } return nil, errors.Errorf("invalid field type: '%s'", o.Type) } diff --git a/holder.go b/holder.go index 9016b9c73..ff27d7b39 100644 --- a/holder.go +++ b/holder.go @@ -1495,7 +1495,7 @@ func (s *holderSyncer) setTranslateReadOnlyFlags(snap *disco.ClusterSnapshot) { } for _, field := range index.Fields() { - if !strings.EqualFold(field.options.Type, FieldTypeVarchar) { + if !(strings.EqualFold(field.options.Type, FieldTypeVarchar) || strings.EqualFold(field.options.Type, FieldTypeVector)) { field.TranslateStore().SetReadOnly(!isPrimaryFieldTranslator) } } diff --git a/http_handler.go b/http_handler.go index 9f89beb82..b7dd0d3cf 100644 --- a/http_handler.go +++ b/http_handler.go @@ -2074,6 +2074,8 @@ func fieldOptionsToFunctionalOpts(opt fieldOptions) []FieldOption { fos = append(fos, OptFieldTypeBool()) case FieldTypeVarchar: fos = append(fos, OptFieldTypeVarchar(*opt.Length)) + case FieldTypeVector: + fos = append(fos, OptFieldTypeVector(*opt.Length)) } if opt.Keys != nil { if *opt.Keys { @@ -2316,6 +2318,10 @@ func (o *fieldOptions) validate() error { if o.Length == nil { return NewBadRequestError(errors.New("varchar field requires a length argument")) } + case FieldTypeVector: + if o.Length == nil { + return NewBadRequestError(errors.New("vector field requires a length argument")) + } default: return errors.Errorf("invalid field type: %s", o.Type) } diff --git a/index.go b/index.go index 99496720f..57b065185 100644 --- a/index.go +++ b/index.go @@ -383,7 +383,7 @@ func (i *Index) openField(mu *sync.Mutex, cfm *CreateFieldMessage, file string) var err error var fld *Field mu.Lock() - if strings.EqualFold(cfm.Meta.Type, FieldTypeVarchar) { + if strings.EqualFold(cfm.Meta.Type, FieldTypeVarchar) || strings.EqualFold(cfm.Meta.Type, FieldTypeVector) { fld, err = i.newNonRBFField(i.fieldPath(filepath.Base(file)), filepath.Base(file)) } else { fld, err = i.newField(i.fieldPath(filepath.Base(file)), filepath.Base(file)) @@ -467,7 +467,7 @@ func (i *Index) setFieldBitDepths() error { func (i *Index) hasTStoreFields() bool { for _, f := range i.fields { switch f.Type() { - case FieldTypeVarchar: + case FieldTypeVarchar, FieldTypeVector: return true } } @@ -978,7 +978,7 @@ func (i *Index) createField(cfm *CreateFieldMessage) (*Field, error) { var err error var f *Field // initialize non-rbf field - if strings.EqualFold(cfm.Meta.Type, FieldTypeVarchar) { + if strings.EqualFold(cfm.Meta.Type, FieldTypeVarchar) || strings.EqualFold(cfm.Meta.Type, FieldTypeVector) { f, err = i.newNonRBFField(i.fieldPath(cfm.Field), cfm.Field) if err != nil { return nil, errors.Wrap(err, "initializing") @@ -1118,7 +1118,7 @@ func (i *Index) GetTStore(shard uint64) (*tstore.BTree, error) { fieldList := make([]*Field, 0) for _, f := range i.fields { - if strings.EqualFold(f.options.Type, FieldTypeVarchar) { + if strings.EqualFold(f.options.Type, FieldTypeVarchar) || strings.EqualFold(f.options.Type, FieldTypeVector) { fieldList = append(fieldList, f) } } @@ -1129,9 +1129,19 @@ func (i *Index) GetTStore(shard uint64) (*tstore.BTree, error) { indexSchema := make(planner_types.Schema, len(fieldList)) for i, f := range fieldList { - indexSchema[i] = &planner_types.PlannerColumn{ - ColumnName: f.name, - Type: parser.NewDataTypeVarchar(f.options.Length), + switch f.Type() { + case FieldTypeVarchar: + indexSchema[i] = &planner_types.PlannerColumn{ + ColumnName: f.name, + Type: parser.NewDataTypeVarchar(f.options.Length), + } + case FieldTypeVector: + indexSchema[i] = &planner_types.PlannerColumn{ + ColumnName: f.name, + Type: parser.NewDataTypeVarchar(f.options.Length), + } + default: + return nil, errors.Errorf("unexepected field type '%s'", f.Type()) } } diff --git a/schema.go b/schema.go index 52db8475c..7ccc1fb95 100644 --- a/schema.go +++ b/schema.go @@ -249,6 +249,9 @@ func FieldInfoToField(fi *FieldInfo) *dax.Field { case FieldTypeVarchar: fieldType = dax.BaseTypeVarchar length = fo.Length + case FieldTypeVector: + fieldType = dax.BaseTypeVector + length = fo.Length default: panic(fmt.Sprintf("unhandled featurebase field type: %s", fo.Type)) } @@ -501,6 +504,10 @@ func FieldOptionsFromField(fld *dax.Field) ([]FieldOption, error) { opts = append(opts, OptFieldTypeVarchar(fld.Options.Length), ) + case dax.BaseTypeVector: + opts = append(opts, + OptFieldTypeVector(fld.Options.Length), + ) default: return nil, errors.Errorf("unsupport field type: %s", fld.Type) diff --git a/sql3/errors.go b/sql3/errors.go index 56495ecc6..0bd85be3f 100644 --- a/sql3/errors.go +++ b/sql3/errors.go @@ -48,6 +48,7 @@ const ( // varchar ErrVarcharLengthExpected errors.Code = "ErrVarcharLengthExpected" + ErrVectorLengthExpected errors.Code = "ErrVectorLengthExpected" ErrInvalidCast errors.Code = "ErrInvalidCast" ErrInvalidTypeCoercion errors.Code = "ErrInvalidTypeCoercion" @@ -517,6 +518,13 @@ func NewErrVarcharLengthExpected(line, col int) error { ) } +func NewErrVectorLengthExpected(line, col int) error { + return errors.New( + ErrVectorLengthExpected, + fmt.Sprintf("[%d:%d] vector length expected", line, col), + ) +} + func NewErrInvalidTimeUnit(line, col int, unit string) error { return errors.New( ErrInvalidTimeUnit, diff --git a/sql3/parser/ast.go b/sql3/parser/ast.go index 47e7f98b0..7ff263012 100644 --- a/sql3/parser/ast.go +++ b/sql3/parser/ast.go @@ -77,7 +77,7 @@ func (*OrderingTerm) node() {} func (*OverClause) node() {} func (*ParenExpr) node() {} func (*PredictStatement) node() {} -func (*SetLiteralExpr) node() {} +func (*ArrayLiteralExpr) node() {} func (*ParenSource) node() {} func (*PrimaryKeyConstraint) node() {} func (*QualifiedRef) node() {} @@ -269,7 +269,7 @@ func (*NullLit) expr() {} func (*IntegerLit) expr() {} func (*FloatLit) expr() {} func (*ParenExpr) expr() {} -func (*SetLiteralExpr) expr() {} +func (*ArrayLiteralExpr) expr() {} func (*TupleLiteralExpr) expr() {} func (*QualifiedRef) expr() {} func (*Range) expr() {} @@ -324,7 +324,7 @@ func CloneExpr(expr Expr) Expr { return expr.Clone() case *DateLit: return expr.Clone() - case *SetLiteralExpr: + case *ArrayLiteralExpr: return expr.Clone() default: panic(fmt.Sprintf("invalid expr type: %T", expr)) @@ -4701,7 +4701,7 @@ func (expr *ParenExpr) String() string { return fmt.Sprintf("(%s)", expr.X.String()) } -type SetLiteralExpr struct { +type ArrayLiteralExpr struct { Lbracket Pos // position of left bracket Members []Expr // bracketed expression Rbracket Pos // position of right bracket @@ -4709,20 +4709,20 @@ type SetLiteralExpr struct { ResultDataType ExprDataType } -func (expr *SetLiteralExpr) IsLiteral() bool { +func (expr *ArrayLiteralExpr) IsLiteral() bool { return true } -func (expr *SetLiteralExpr) DataType() ExprDataType { +func (expr *ArrayLiteralExpr) DataType() ExprDataType { return expr.ResultDataType } -func (expr *SetLiteralExpr) Pos() Pos { +func (expr *ArrayLiteralExpr) Pos() Pos { return expr.Lbracket } // Clone returns a deep copy of expr. -func (expr *SetLiteralExpr) Clone() *SetLiteralExpr { +func (expr *ArrayLiteralExpr) Clone() *ArrayLiteralExpr { if expr == nil { return nil } @@ -4732,7 +4732,7 @@ func (expr *SetLiteralExpr) Clone() *SetLiteralExpr { } // String returns the string representation of the expression. -func (expr *SetLiteralExpr) String() string { +func (expr *ArrayLiteralExpr) String() string { var buf bytes.Buffer if len(expr.Members) != 0 { diff --git a/sql3/parser/ast_test.go b/sql3/parser/ast_test.go index 236adfb30..406e8596f 100644 --- a/sql3/parser/ast_test.go +++ b/sql3/parser/ast_test.go @@ -1353,7 +1353,7 @@ func TestDateLit_String(t *testing.T) { // test SetLiteralExpr. func TestSetLiteralExpr_String(t *testing.T) { - sl := &parser.SetLiteralExpr{ + sl := &parser.ArrayLiteralExpr{ Lbracket: pos(0), Rbracket: pos(0), Members: []parser.Expr{ diff --git a/sql3/parser/astdatatype.go b/sql3/parser/astdatatype.go index ff0ebf6dd..89894f35b 100644 --- a/sql3/parser/astdatatype.go +++ b/sql3/parser/astdatatype.go @@ -19,7 +19,8 @@ func IsValidTypeName(typeName string) bool { dax.BaseTypeStringSet, dax.BaseTypeStringSetQ, dax.BaseTypeTimestamp, - dax.BaseTypeVarchar: + dax.BaseTypeVarchar, + dax.BaseTypeVector: return true default: return false @@ -42,9 +43,10 @@ type ExprDataType interface { } func (*DataTypeVoid) exprDataType() {} +func (*DataTypeArray) exprDataType() {} func (*DataTypeRange) exprDataType() {} func (*DataTypeTuple) exprDataType() {} -func (*DataTypeSubtable) exprDataType() {} +func (*DataTypeSubtable) exprDataType() {} // TODO (pok) don't think we need this one? func (*DataTypeBool) exprDataType() {} func (*DataTypeDecimal) exprDataType() {} func (*DataTypeID) exprDataType() {} @@ -56,6 +58,7 @@ func (*DataTypeStringSet) exprDataType() {} func (*DataTypeStringSetQuantum) exprDataType() {} func (*DataTypeTimestamp) exprDataType() {} func (*DataTypeVarchar) exprDataType() {} +func (*DataTypeVector) exprDataType() {} func (*DataTypeVarbinary) exprDataType() {} type DataTypeVoid struct { @@ -77,6 +80,28 @@ func (*DataTypeVoid) TypeInfo() map[string]interface{} { return nil } +type DataTypeArray struct { + SubscriptType ExprDataType +} + +func NewDataTypeArray(subscriptType ExprDataType) *DataTypeArray { + return &DataTypeArray{ + SubscriptType: subscriptType, + } +} + +func (dt *DataTypeArray) BaseTypeName() string { + return "array" +} + +func (dt *DataTypeArray) TypeDescription() string { + return fmt.Sprintf("array(%s)", dt.SubscriptType.TypeDescription()) +} + +func (*DataTypeArray) TypeInfo() map[string]interface{} { + return nil +} + type DataTypeRange struct { SubscriptType ExprDataType } @@ -229,6 +254,30 @@ func (d *DataTypeVarchar) TypeInfo() map[string]interface{} { } } +type DataTypeVector struct { + Length int64 +} + +func NewDataTypeVector(length int64) *DataTypeVector { + return &DataTypeVector{ + Length: length, + } +} + +func (d *DataTypeVector) BaseTypeName() string { + return dax.BaseTypeVector +} + +func (d *DataTypeVector) TypeDescription() string { + return fmt.Sprintf("%s(%d)", dax.BaseTypeVector, d.Length) +} + +func (d *DataTypeVector) TypeInfo() map[string]interface{} { + return map[string]interface{}{ + "length": d.Length, + } +} + type DataTypeVarbinary struct { Length int64 } diff --git a/sql3/parser/parser.go b/sql3/parser/parser.go index 6f3990cfd..3637d70fa 100644 --- a/sql3/parser/parser.go +++ b/sql3/parser/parser.go @@ -3475,8 +3475,8 @@ func (p *Parser) parseParenExpr() (_ *ParenExpr, err error) { return &expr, nil } -func (p *Parser) parseSetLiteralExpr() (_ *SetLiteralExpr, err error) { - var expr SetLiteralExpr +func (p *Parser) parseSetLiteralExpr() (_ *ArrayLiteralExpr, err error) { + var expr ArrayLiteralExpr expr.Lbracket, _, _ = p.scan() for p.peek() != RB { diff --git a/sql3/parser/token.go b/sql3/parser/token.go index 165e07243..b15f8be82 100644 --- a/sql3/parser/token.go +++ b/sql3/parser/token.go @@ -244,7 +244,6 @@ const ( USING VACUUM VALUES - VECTOR VIEW VIRTUAL WHEN @@ -471,7 +470,6 @@ var tokens = [...]string{ USING: "USING", VACUUM: "VACUUM", VALUES: "VALUES", - VECTOR: "VECTOR", VIEW: "VIEW", VIRTUAL: "VIRTUAL", WHEN: "WHEN", diff --git a/sql3/planner/compilecreatemodel.go b/sql3/planner/compilecreatemodel.go index 9cd57a056..42bb04e87 100644 --- a/sql3/planner/compilecreatemodel.go +++ b/sql3/planner/compilecreatemodel.go @@ -43,7 +43,7 @@ func (p *ExecutionPlanner) compileCreateModelStatement(stmt *parser.CreateModelS model.modelType = lit.Value case "labels": - lit, ok := o.OptionExpr.(*parser.SetLiteralExpr) + lit, ok := o.OptionExpr.(*parser.ArrayLiteralExpr) if !ok { return nil, sql3.NewErrInternalf("unexpected type '%T'", o.OptionExpr) } diff --git a/sql3/planner/compilecreatetable.go b/sql3/planner/compilecreatetable.go index 4d2ef91e4..7a7532ca5 100644 --- a/sql3/planner/compilecreatetable.go +++ b/sql3/planner/compilecreatetable.go @@ -249,6 +249,20 @@ func (p *ExecutionPlanner) compileColumn(ctx context.Context, col *parser.Column } column.fos = append(column.fos, pilosa.OptFieldTypeVarchar(length)) + + case dax.BaseTypeVector: + // if we don't have a length, it's an error + if col.Type.Modifier == nil { + return nil, sql3.NewErrVectorLengthExpected(col.Type.Name.NamePos.Line, col.Type.Name.NamePos.Column) + } + + // get the modifier value + length, err = strconv.ParseInt(col.Type.Modifier.Value, 10, 64) + if err != nil { + return nil, err + } + + column.fos = append(column.fos, pilosa.OptFieldTypeVector(length)) } return column, nil } diff --git a/sql3/planner/expression.go b/sql3/planner/expression.go index 721ff86e3..d54b9c54b 100644 --- a/sql3/planner/expression.go +++ b/sql3/planner/expression.go @@ -2435,22 +2435,27 @@ func (n *exprListPlanExpression) WithChildren(children ...types.PlanExpression) return newExprListExpression(children), nil } -// exprSetLiteralPlanExpression is a set literal -type exprSetLiteralPlanExpression struct { +// exprArrayLiteralPlanExpression is a set literal +type exprArrayLiteralPlanExpression struct { members []types.PlanExpression dataType parser.ExprDataType } -func newExprSetLiteralPlanExpression(members []types.PlanExpression, dataType parser.ExprDataType) *exprSetLiteralPlanExpression { - return &exprSetLiteralPlanExpression{ +func newExprArrayLiteralPlanExpression(members []types.PlanExpression, dataType parser.ExprDataType) *exprArrayLiteralPlanExpression { + return &exprArrayLiteralPlanExpression{ members: members, dataType: dataType, } } -func (n *exprSetLiteralPlanExpression) Evaluate(currentRow []interface{}) (interface{}, error) { - switch typ := n.dataType.(type) { - case *parser.DataTypeIDSet: +func (n *exprArrayLiteralPlanExpression) Evaluate(currentRow []interface{}) (interface{}, error) { + arrayType, ok := n.dataType.(*parser.DataTypeArray) + if !ok { + return nil, sql3.NewErrInternalf("unexepcted array literal type '%T'", n.dataType) + } + + switch typ := arrayType.SubscriptType.(type) { + case *parser.DataTypeID: result := []int64{} for _, e := range n.members { er, err := e.Evaluate(currentRow) @@ -2469,7 +2474,7 @@ func (n *exprSetLiteralPlanExpression) Evaluate(currentRow []interface{}) (inter } return result, nil - case *parser.DataTypeStringSet: + case *parser.DataTypeString: result := []string{} for _, e := range n.members { er, err := e.Evaluate(currentRow) @@ -2483,16 +2488,32 @@ func (n *exprSetLiteralPlanExpression) Evaluate(currentRow []interface{}) (inter result = append(result, ers) } return result, nil + + case *parser.DataTypeDecimal: + result := []pql.Decimal{} + for _, e := range n.members { + er, err := e.Evaluate(currentRow) + if err != nil { + return nil, err + } + ers, ok := er.(pql.Decimal) + if !ok { + return nil, sql3.NewErrInternalf("unable to convert element result") + } + result = append(result, ers) + } + return result, nil + default: - return nil, sql3.NewErrInternalf("unexpected set literal type '%T'", typ) + return nil, sql3.NewErrInternalf("unexpected array literal subscript type '%T'", typ) } } -func (n *exprSetLiteralPlanExpression) Type() parser.ExprDataType { +func (n *exprArrayLiteralPlanExpression) Type() parser.ExprDataType { return n.dataType } -func (n *exprSetLiteralPlanExpression) String() string { +func (n *exprArrayLiteralPlanExpression) String() string { var members string for idx, m := range n.members { if idx > 0 { @@ -2503,7 +2524,7 @@ func (n *exprSetLiteralPlanExpression) String() string { return fmt.Sprintf("[%s]", members) } -func (n *exprSetLiteralPlanExpression) Plan() map[string]interface{} { +func (n *exprArrayLiteralPlanExpression) Plan() map[string]interface{} { result := make(map[string]interface{}) result["_expr"] = fmt.Sprintf("%T", n) result["description"] = n.String() @@ -2515,15 +2536,15 @@ func (n *exprSetLiteralPlanExpression) Plan() map[string]interface{} { return result } -func (n *exprSetLiteralPlanExpression) Children() []types.PlanExpression { +func (n *exprArrayLiteralPlanExpression) Children() []types.PlanExpression { return n.members } -func (n *exprSetLiteralPlanExpression) WithChildren(children ...types.PlanExpression) (types.PlanExpression, error) { +func (n *exprArrayLiteralPlanExpression) WithChildren(children ...types.PlanExpression) (types.PlanExpression, error) { if len(children) != len(n.members) { return nil, sql3.NewErrInternalf("unexpected number of children '%d'", len(children)) } - return newExprSetLiteralPlanExpression(children, n.dataType), nil + return newExprArrayLiteralPlanExpression(children, n.dataType), nil } // exprTupleLiteralPlanExpression is a tuple literal @@ -2630,7 +2651,7 @@ func (p *ExecutionPlanner) compileExpr(expr parser.Expr) (_ types.PlanExpression } return newExprListExpression(exprList), nil - case *parser.SetLiteralExpr: + case *parser.ArrayLiteralExpr: exprList := []types.PlanExpression{} for _, e := range expr.Members { listExpr, err := p.compileExpr(e) @@ -2639,7 +2660,7 @@ func (p *ExecutionPlanner) compileExpr(expr parser.Expr) (_ types.PlanExpression } exprList = append(exprList, listExpr) } - return newExprSetLiteralPlanExpression(exprList, expr.DataType()), nil + return newExprArrayLiteralPlanExpression(exprList, expr.DataType()), nil case *parser.TupleLiteralExpr: exprList := []types.PlanExpression{} diff --git a/sql3/planner/expression_it_test.go b/sql3/planner/expression_it_test.go index d9f8d870b..81e1c2091 100644 --- a/sql3/planner/expression_it_test.go +++ b/sql3/planner/expression_it_test.go @@ -77,7 +77,7 @@ func TestExpressions(t *testing.T) { elop := newExprListExpression([]types.PlanExpression{newStringLiteralPlanExpression("foo"), newStringLiteralPlanExpression("bar")}) assert.Equal(t, elop.String(), "('foo', 'bar')") - stlop := newExprSetLiteralPlanExpression([]types.PlanExpression{newStringLiteralPlanExpression("foo"), newStringLiteralPlanExpression("bar")}, parser.NewDataTypeString()) + stlop := newExprArrayLiteralPlanExpression([]types.PlanExpression{newStringLiteralPlanExpression("foo"), newStringLiteralPlanExpression("bar")}, parser.NewDataTypeString()) assert.Equal(t, stlop.String(), "['foo', 'bar']") tplop := newExprTupleLiteralPlanExpression([]types.PlanExpression{newStringLiteralPlanExpression("foo"), newStringLiteralPlanExpression("bar")}, parser.NewDataTypeString()) diff --git a/sql3/planner/expressionanalyzer.go b/sql3/planner/expressionanalyzer.go index 76ce2068a..6e6a5c36a 100644 --- a/sql3/planner/expressionanalyzer.go +++ b/sql3/planner/expressionanalyzer.go @@ -162,7 +162,7 @@ func (p *ExecutionPlanner) analyzeExpression(ctx context.Context, expr parser.Ex e.X = pexpr return e, nil - case *parser.SetLiteralExpr: + case *parser.ArrayLiteralExpr: for i, ex := range e.Members { listExpr, err := p.analyzeExpression(ctx, ex, scope) if err != nil { @@ -175,8 +175,8 @@ func (p *ExecutionPlanner) analyzeExpression(ctx context.Context, expr parser.Ex return nil, sql3.NewErrLiteralEmptySetNotAllowed(e.Lbracket.Line, e.Lbracket.Column) } - setDataType := e.Members[0].DataType() - switch setDataType.(type) { + subDataType := e.Members[0].DataType() + switch subDataType.(type) { case *parser.DataTypeID, *parser.DataTypeInt: //make sure everything else is an int for _, mbr := range e.Members { @@ -184,7 +184,6 @@ func (p *ExecutionPlanner) analyzeExpression(ctx context.Context, expr parser.Ex return nil, sql3.NewErrIntExpressionExpected(mbr.Pos().Line, mbr.Pos().Column) } } - e.ResultDataType = parser.NewDataTypeIDSet() case *parser.DataTypeString: //make sure everything else is a string @@ -193,12 +192,20 @@ func (p *ExecutionPlanner) analyzeExpression(ctx context.Context, expr parser.Ex return nil, sql3.NewErrStringExpressionExpected(mbr.Pos().Line, mbr.Pos().Column) } } - e.ResultDataType = parser.NewDataTypeStringSet() + + case *parser.DataTypeDecimal: + //make sure everything else is a decimal + for _, mbr := range e.Members { + if !typeIsDecimal(mbr.DataType()) { + return nil, sql3.NewErrInternalf("unexpected data type") + } + } default: return nil, sql3.NewErrSetLiteralMustContainIntOrString(e.Members[0].Pos().Line, e.Members[0].Pos().Column) } + e.ResultDataType = parser.NewDataTypeArray(subDataType) return e, nil case *parser.TupleLiteralExpr: diff --git a/sql3/planner/expressionpql.go b/sql3/planner/expressionpql.go index 7d19e3ac7..af165003c 100644 --- a/sql3/planner/expressionpql.go +++ b/sql3/planner/expressionpql.go @@ -43,7 +43,7 @@ func (p *ExecutionPlanner) generatePQLCallFromExpr(ctx context.Context, expr typ case "SETCONTAINSALL": col := expr.args[0].(*qualifiedRefPlanExpression) - set, ok := expr.args[1].(*exprSetLiteralPlanExpression) + set, ok := expr.args[1].(*exprArrayLiteralPlanExpression) if !ok { return nil, sql3.NewErrInternalf("unexpected argument type '%T'", expr.args[1]) } @@ -71,7 +71,7 @@ func (p *ExecutionPlanner) generatePQLCallFromExpr(ctx context.Context, expr typ case "SETCONTAINSANY": col := expr.args[0].(*qualifiedRefPlanExpression) - set, ok := expr.args[1].(*exprSetLiteralPlanExpression) + set, ok := expr.args[1].(*exprArrayLiteralPlanExpression) if !ok { return nil, sql3.NewErrInternalf("unexpected argument type '%T'", expr.args[1]) } diff --git a/sql3/planner/expressiontypes.go b/sql3/planner/expressiontypes.go index 3d2df6b18..cea52a895 100644 --- a/sql3/planner/expressiontypes.go +++ b/sql3/planner/expressiontypes.go @@ -71,6 +71,8 @@ func FieldSQLDataType(f *pilosa.FieldInfo) parser.ExprDataType { case pilosa.FieldTypeVarchar: return parser.NewDataTypeVarchar(f.Options.Length) + case pilosa.FieldTypeVector: + return parser.NewDataTypeVector(f.Options.Length) default: return parser.NewDataTypeVoid() } @@ -249,6 +251,19 @@ func typesAreAssignmentCompatible(targetType parser.ExprDataType, sourceType par switch lhs := targetType.(type) { + case *parser.DataTypeVector: + switch st := sourceType.(type) { + case *parser.DataTypeArray: + switch st.SubscriptType.(type) { + case *parser.DataTypeDecimal: + return true + default: + return false + } + default: + return false + } + case *parser.DataTypeInt: switch sourceType.(type) { case *parser.DataTypeInt: diff --git a/sql3/planner/opbulkinsert.go b/sql3/planner/opbulkinsert.go index 2586ad34b..f5a4a1072 100644 --- a/sql3/planner/opbulkinsert.go +++ b/sql3/planner/opbulkinsert.go @@ -849,7 +849,7 @@ func processColumnValue(rawValue interface{}, targetType parser.ExprDataType) (t for _, m := range val { members = append(members, newIntLiteralPlanExpression(m)) } - return newExprSetLiteralPlanExpression(members, parser.NewDataTypeIDSet()), nil + return newExprArrayLiteralPlanExpression(members, parser.NewDataTypeArray(parser.NewDataTypeID())), nil case *parser.DataTypeStringSet: val, ok := rawValue.([]string) @@ -860,7 +860,7 @@ func processColumnValue(rawValue interface{}, targetType parser.ExprDataType) (t for _, m := range val { members = append(members, newStringLiteralPlanExpression(m)) } - return newExprSetLiteralPlanExpression(members, parser.NewDataTypeStringSet()), nil + return newExprArrayLiteralPlanExpression(members, parser.NewDataTypeArray(parser.NewDataTypeString())), nil case *parser.DataTypeTimestamp: tval, ok := rawValue.(time.Time) diff --git a/sql3/planner/opcopy.go b/sql3/planner/opcopy.go index 2ede81ff0..6a4c0c05b 100644 --- a/sql3/planner/opcopy.go +++ b/sql3/planner/opcopy.go @@ -227,7 +227,7 @@ func (i *copyIterator) Next(ctx context.Context) (types.Row, error) { for _, m := range val { members = append(members, newStringLiteralPlanExpression(m)) } - irow[i] = newExprSetLiteralPlanExpression(members, parser.NewDataTypeStringSet()) + irow[i] = newExprArrayLiteralPlanExpression(members, parser.NewDataTypeArray(parser.NewDataTypeString())) case *parser.DataTypeIDSet: val, ok := row[i].([]int64) @@ -239,7 +239,7 @@ func (i *copyIterator) Next(ctx context.Context) (types.Row, error) { for _, m := range val { members = append(members, newIntLiteralPlanExpression(m)) } - irow[i] = newExprSetLiteralPlanExpression(members, parser.NewDataTypeIDSet()) + irow[i] = newExprArrayLiteralPlanExpression(members, parser.NewDataTypeArray(parser.NewDataTypeID())) default: return nil, sql3.NewErrInternalf("unhandled type '%T'", ty) diff --git a/sql3/planner/opinsert.go b/sql3/planner/opinsert.go index b7e86b346..22f7f1e1c 100644 --- a/sql3/planner/opinsert.go +++ b/sql3/planner/opinsert.go @@ -429,6 +429,18 @@ func (i *insertRowIter) Next(ctx context.Context) (types.Row, error) { return nil, sql3.NewErrInternalf("unexpected varchar type '%T'", v) } + case pilosa.FieldTypeVector: + switch v := eval.(type) { + case []pql.Decimal: + nv := make([]float64, len(v)) + for j, vv := range v { + nv[j] = vv.Float64() + } + row.Values[posVals[idx]] = nv + default: + return nil, sql3.NewErrInternalf("unexpected vector type '%T'", v) + } + default: row.Values[posVals[idx]] = eval } diff --git a/tstore/btree.go b/tstore/btree.go index 6ac501422..38a658a49 100644 --- a/tstore/btree.go +++ b/tstore/btree.go @@ -116,6 +116,9 @@ func NewBTree(maxKeySize int, objectID int32, shard int32, schema types.Schema, case *parser.DataTypeVarchar: payLoadLength += 4 // offset or null payLoadLength += 4 + int(ty.Length) // actual data + case *parser.DataTypeVector: + payLoadLength += 4 // offset or null + payLoadLength += 4 + int(ty.Length)*8 // actual data default: return nil, errors.Errorf("unsupported t-store data type '%T'", ty) } diff --git a/tstore/btreetypes.go b/tstore/btreetypes.go index 54049dd79..a2ef8bc92 100644 --- a/tstore/btreetypes.go +++ b/tstore/btreetypes.go @@ -6,6 +6,7 @@ import ( "bytes" "encoding/binary" "io" + "math" "github.com/pkg/errors" @@ -118,6 +119,19 @@ func NewBTreeTupleFromBytes(b []byte, schema types.Schema) *BTreeTuple { binary.Read(dataRdr, binary.BigEndian, &bvalue) t.Tuple[i] = string(bvalue) + case *parser.DataTypeVector: + dataRdr.Seek(int64(fieldOffset), io.SeekStart) + var l int32 + binary.Read(dataRdr, binary.BigEndian, &l) + bvalue := make([]float64, l) + + for j := 0; j < int(l); j++ { + var fvalue float64 + binary.Read(dataRdr, binary.BigEndian, &fvalue) + bvalue[j] = fvalue + } + t.Tuple[i] = bvalue + case *parser.DataTypeVarbinary: dataRdr.Seek(int64(fieldOffset), io.SeekStart) var l int32 @@ -204,11 +218,32 @@ func (b *BTreeTuple) Bytes(tid TID, schema types.Schema, schemaVersion int, vers data := rd.(string) b = make([]byte, 4) l := len(data) + // update the offset to be the end of the data we're about to write offset += 4 + l binary.BigEndian.PutUint32(b, uint32(l)) fieldData.Write(b) fieldData.WriteString(data) + case *parser.DataTypeVector: + // write field offset + b := make([]byte, 4) + binary.BigEndian.PutUint32(b, uint32(offset)) + valueBuf.Write(b) + + // write field data + data := rd.([]float64) + b = make([]byte, 4) + l := len(data) + // update the offset to be the end of the data we're about to write + offset += 4 + l*8 + binary.BigEndian.PutUint32(b, uint32(l)) + fieldData.Write(b) + for _, f := range data { + var fbuf [8]byte + binary.BigEndian.PutUint64(fbuf[:], math.Float64bits(f)) + fieldData.Write(fbuf[:]) + } + case *parser.DataTypeVarbinary: // write field offset b := make([]byte, 4) @@ -219,6 +254,7 @@ func (b *BTreeTuple) Bytes(tid TID, schema types.Schema, schemaVersion int, vers data := rd.([]byte) b = make([]byte, 4) l := len(data) + // update the offset to be the end of the data we're about to write offset += 4 + l binary.BigEndian.PutUint32(b, uint32(l)) binary.BigEndian.PutUint32(b, uint32(len(data))) diff --git a/wireprotocol/wireprimitives.go b/wireprotocol/wireprimitives.go index 8591cfaac..b67c32167 100644 --- a/wireprotocol/wireprimitives.go +++ b/wireprotocol/wireprimitives.go @@ -5,6 +5,7 @@ import ( "bytes" "encoding/binary" "io" + "math" "time" "github.com/featurebasedb/featurebase/v3/errors" @@ -37,7 +38,8 @@ const ( TYPE_STRING int8 = 0x07 TYPE_STRINGSET int8 = 0x08 TYPE_VARCHAR int8 = 0x09 - TYPE_VARBINARY int8 = 0xA + TYPE_VARBINARY int8 = 0x0A + TYPE_VECTOR int8 = 0x0B ) func ExpectToken(reader io.Reader, token int16) (int16, error) { @@ -124,6 +126,10 @@ func WriteSchema(schema types.Schema) ([]byte, error) { writeInt8(writer, TYPE_VARCHAR) writeInt32(writer, int32(ty.Length)) + case *parser.DataTypeVector: + writeInt8(writer, TYPE_VECTOR) + writeInt32(writer, int32(ty.Length)) + case *parser.DataTypeVarbinary: writeInt8(writer, TYPE_VARBINARY) writeInt32(writer, int32(ty.Length)) @@ -207,6 +213,14 @@ func ReadSchema(reader io.Reader) (types.Schema, error) { } dataType = parser.NewDataTypeVarchar(int64(length)) + case TYPE_VECTOR: + var length int32 + err = binary.Read(reader, binary.BigEndian, &length) + if err != nil { + return nil, err + } + dataType = parser.NewDataTypeVector(int64(length)) + case TYPE_VARBINARY: var length int32 err = binary.Read(reader, binary.BigEndian, &length) @@ -385,6 +399,22 @@ func WriteRow(row types.Row, schema types.Schema) ([]byte, error) { writer.WriteString(v) } + case *parser.DataTypeVector: + if val == nil { + writeInt32(writer, 0) + } else { + v, ok := row[i].([]float64) + if !ok { + return []byte{}, errors.Errorf("unexpected type '%T'", row[i]) + } + writeInt32(writer, int32(len(v))) + for _, f := range v { + var fbuf [8]byte + binary.BigEndian.PutUint64(fbuf[:], math.Float64bits(f)) + writer.Write(fbuf[:]) + } + } + case *parser.DataTypeVarbinary: if val == nil { writeInt32(writer, 0) @@ -559,6 +589,28 @@ func ReadRow(reader io.Reader, schema types.Schema) (types.Row, error) { row[idx] = string(bvalue) } + case *parser.DataTypeVector: + var len int32 + err := binary.Read(reader, binary.BigEndian, &len) + if err != nil { + return nil, err + } + if len == 0 { + row[idx] = nil + } else { + bvalue := make([]float64, len) + for j := 0; j < int(len); j++ { + var fvalue float64 + err = binary.Read(reader, binary.BigEndian, &fvalue) + if err != nil { + return nil, err + } + bvalue[j] = fvalue + } + + row[idx] = bvalue + } + case *parser.DataTypeVarbinary: var len int32 err := binary.Read(reader, binary.BigEndian, &len)