mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
a158bba125
We had a bunch of different places which had basically the same logic, except that some were testing both RFC3339Nano and RFC3339 formats, and some weren't. This turns out not to matter, because the fractional second part is always permitted and never required, so those two formats are identical. Mostly, though, we now ensure that everything we do that is trying to convert timestamps has the same logic, so if we want to make changes to that logic, we have a central point, which lives in the parser. This came out of an attempt to figure out why the RFC3339 case wasn't getting any test coverage.
453 lines
15 KiB
Go
453 lines
15 KiB
Go
// Copyright 2021 Molecula Corp. All rights reserved.
|
|
|
|
package planner
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
pilosa "github.com/featurebasedb/featurebase/v3"
|
|
fbbatch "github.com/featurebasedb/featurebase/v3/batch"
|
|
"github.com/featurebasedb/featurebase/v3/dax"
|
|
"github.com/featurebasedb/featurebase/v3/pql"
|
|
"github.com/featurebasedb/featurebase/v3/sql3"
|
|
"github.com/featurebasedb/featurebase/v3/sql3/parser"
|
|
"github.com/featurebasedb/featurebase/v3/sql3/planner/types"
|
|
"github.com/pkg/errors"
|
|
)
|
|
|
|
// PlanOpInsert plan operator to handle INSERT.
|
|
type PlanOpInsert struct {
|
|
planner *ExecutionPlanner
|
|
tableName string
|
|
targetColumns []*qualifiedRefPlanExpression
|
|
insertValues [][]types.PlanExpression
|
|
warnings []string
|
|
}
|
|
|
|
func NewPlanOpInsert(p *ExecutionPlanner, tableName string, targetColumns []*qualifiedRefPlanExpression, insertValues [][]types.PlanExpression) *PlanOpInsert {
|
|
return &PlanOpInsert{
|
|
planner: p,
|
|
tableName: tableName,
|
|
targetColumns: targetColumns,
|
|
insertValues: insertValues,
|
|
warnings: make([]string, 0),
|
|
}
|
|
}
|
|
|
|
func (p *PlanOpInsert) Plan() map[string]interface{} {
|
|
result := make(map[string]interface{})
|
|
result["_op"] = fmt.Sprintf("%T", p)
|
|
result["_schema"] = p.Schema().Plan()
|
|
result["tableName"] = p.tableName
|
|
ps := make([]interface{}, 0)
|
|
for _, e := range p.targetColumns {
|
|
ps = append(ps, e.Plan())
|
|
}
|
|
result["targetColumns"] = ps
|
|
result["insertTupleCount"] = len(p.insertValues)
|
|
return result
|
|
}
|
|
|
|
func (p *PlanOpInsert) String() string {
|
|
return ""
|
|
}
|
|
|
|
func (p *PlanOpInsert) AddWarning(warning string) {
|
|
p.warnings = append(p.warnings, warning)
|
|
}
|
|
|
|
func (p *PlanOpInsert) Warnings() []string {
|
|
return p.warnings
|
|
}
|
|
|
|
func (p *PlanOpInsert) Schema() types.Schema {
|
|
return types.Schema{}
|
|
}
|
|
|
|
func (p *PlanOpInsert) Children() []types.PlanOperator {
|
|
return []types.PlanOperator{}
|
|
}
|
|
|
|
func (p *PlanOpInsert) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) {
|
|
return &insertRowIter{
|
|
planner: p.planner,
|
|
tableName: p.tableName,
|
|
targetColumns: p.targetColumns,
|
|
insertValues: p.insertValues,
|
|
}, nil
|
|
}
|
|
|
|
func (p *PlanOpInsert) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) {
|
|
return NewPlanOpInsert(p.planner, p.tableName, p.targetColumns, p.insertValues), nil
|
|
}
|
|
|
|
type insertRowIter struct {
|
|
planner *ExecutionPlanner
|
|
tableName string
|
|
targetColumns []*qualifiedRefPlanExpression
|
|
insertValues [][]types.PlanExpression
|
|
}
|
|
|
|
var _ types.RowIterator = (*insertRowIter)(nil)
|
|
|
|
func (i *insertRowIter) Next(ctx context.Context) (types.Row, error) {
|
|
// posID is the position of the "_id" column in both the targetColumns and
|
|
// values lists.
|
|
var posID int
|
|
|
|
// posVals maps the position in the tuple to the position in the row.Values.
|
|
// It essentially takes the _id column into account and skips it.
|
|
//
|
|
// So for example, if we have sql:
|
|
// INSERT INTO (a, _id, b, c) VALUES ('aa', 1, 'bb', 'cc');
|
|
//
|
|
// Then we want row.ID = 1 and row.Values to contain {'aa', 'bb', 'cc'}.
|
|
// This means posVals would contain the map []int{0, 1*, 1, 2}, which maps
|
|
// VALUES positions (0,2,3) to row.Values (0, 1, 2). Note, the _id position
|
|
// (shown as 1* in the example above) isn't used because we handle it
|
|
// separately.
|
|
posVals := make([]int, len(i.targetColumns))
|
|
|
|
var foundPosID bool
|
|
for j := range i.targetColumns {
|
|
if foundPosID {
|
|
posVals[j] = j - 1
|
|
continue
|
|
}
|
|
if strings.EqualFold(i.targetColumns[j].columnName, string(dax.PrimaryKeyFieldName)) {
|
|
posID = j
|
|
foundPosID = true
|
|
}
|
|
posVals[j] = j
|
|
}
|
|
|
|
// batchSize is currently set to the size of the entire
|
|
// VALUES list. In the future we may want to break this up into smaller
|
|
// batches.
|
|
batchSize := len(i.insertValues)
|
|
|
|
// idxInfoBase is the full IndexInfo stored in the schema. The instance of
|
|
// IndexInfo used in the import (and created below) will be based on the
|
|
// information from idxInfoBase, but the fields may be a limited subset, and
|
|
// may be in a different order.
|
|
tname := dax.TableName(i.tableName)
|
|
tbl, err := i.planner.schemaAPI.TableByName(ctx, tname)
|
|
if err != nil {
|
|
return nil, sql3.NewErrTableNotFound(0, 0, i.tableName)
|
|
}
|
|
idxInfoBase := pilosa.TableToIndexInfo(tbl)
|
|
|
|
// idxInfo is a subset of idxInfoBase, containing only those fields included
|
|
// in the INSERT INTO statement (i.e. only i.targetcolumns), and in the
|
|
// order specified.
|
|
idxInfo := &pilosa.IndexInfo{
|
|
Name: idxInfoBase.Name,
|
|
CreatedAt: idxInfoBase.CreatedAt,
|
|
Options: idxInfoBase.Options,
|
|
Fields: make([]*pilosa.FieldInfo, len(i.targetColumns)-1),
|
|
ShardWidth: idxInfoBase.ShardWidth,
|
|
}
|
|
|
|
// Set up Fields based on i.targetColumns.
|
|
var counter int
|
|
for ii, targetColumn := range i.targetColumns {
|
|
// Skip the "_id" column.
|
|
if ii == posID {
|
|
continue
|
|
}
|
|
idxInfo.Fields[counter] = idxInfoBase.Field(targetColumn.columnName)
|
|
counter++
|
|
}
|
|
|
|
batch, err := fbbatch.NewBatch(i.planner.importer, batchSize, tbl, idxInfo.Fields,
|
|
fbbatch.OptUseShardTransactionalEndpoint(true),
|
|
)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "setting up batch")
|
|
}
|
|
|
|
var qbatchTime fbbatch.QuantizedTime
|
|
qbatchTime.Set(time.Now().UTC())
|
|
|
|
// row is the single instance of batch.Row allocated. It is re-used
|
|
// throughout the for loop to minimize memory allocation.
|
|
var row fbbatch.Row
|
|
|
|
// Initialize row.Values to the size of the target columns, but exclude the
|
|
// record ID ("_id") since that's stored in row.ID.
|
|
row.Values = make([]interface{}, len(i.targetColumns)-1)
|
|
|
|
for rowNumber, tuple := range i.insertValues {
|
|
// Evaluate and set the record ID.
|
|
if eval, err := tuple[posID].Evaluate(nil); err != nil {
|
|
return nil, errors.Wrapf(err, "evaluating record id: %v", tuple[posID])
|
|
} else {
|
|
// These value types correspond to the types supported in batch.Add().
|
|
switch recid := eval.(type) {
|
|
case string, uint64, []byte:
|
|
row.ID = recid
|
|
case int64:
|
|
if recid < 0 {
|
|
return nil, sql3.NewErrInternalf("_id value cannot be negative: %d", recid)
|
|
}
|
|
row.ID = uint64(recid)
|
|
default:
|
|
// If we get to here, it's likey that the id type is unsupported
|
|
// and will cause an error in batch.Add(). So there's no need to
|
|
// return an error here in this default. The only route here
|
|
// would be if a primary key had a type other than intLiteralPlanExpression
|
|
// or stringLiteralPlanExpression
|
|
|
|
row.ID = eval
|
|
}
|
|
}
|
|
|
|
// Loop over the values in the tuple and populate the Row values.
|
|
for idx, iv := range tuple {
|
|
// Skip the record ID because that was already handled above
|
|
// (prior to this loop).
|
|
if idx == posID {
|
|
continue
|
|
}
|
|
|
|
eval, err := iv.Evaluate(nil)
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "evaluating tuple value: %v", iv)
|
|
}
|
|
|
|
columnName := idxInfo.Fields[posVals[idx]].Name
|
|
|
|
// batch.Add does not typically look at field type to determine how
|
|
// to handle a particular value in a row. Instead, it uses value
|
|
// type. As an example, if the value type is int64, then batch.Add
|
|
// assumes that it should be handled as if going into an `int`
|
|
// field. Therefore, we need to look at the field type here and make
|
|
// sure that the value types being sent through batch.Add align with
|
|
// the field type assumptions that batch.Add is making.
|
|
switch opts := idxInfo.Fields[posVals[idx]].Options; opts.Type {
|
|
|
|
// By the time we get here, we assume that the planner has already
|
|
// determined the value type such that the Evaluate() method called
|
|
// above results in the correct value types. There is one exception:
|
|
// sql3 treats all integer values as int64. This means that an ID
|
|
// field, which expects a uint64 value, would get treated as an int
|
|
// field. In order to avoid that, we cast int64 values to uint64
|
|
// when the field type is set or mutex.
|
|
case pilosa.FieldTypeSet, pilosa.FieldTypeMutex:
|
|
switch v := eval.(type) {
|
|
case int64:
|
|
if v < 0 {
|
|
return nil, sql3.NewErrInternalf("converting negative value to uint64: %d", v)
|
|
}
|
|
row.Values[posVals[idx]] = uint64(v)
|
|
case []int64:
|
|
uint64s := make([]uint64, len(v))
|
|
for i := range v {
|
|
if v[i] < 0 {
|
|
return nil, sql3.NewErrInternalf("converting negative slice value to uint64: %d", v[i])
|
|
}
|
|
uint64s[i] = uint64(v[i])
|
|
}
|
|
row.Values[posVals[idx]] = uint64s
|
|
default:
|
|
row.Values[posVals[idx]] = eval
|
|
}
|
|
|
|
case pilosa.FieldTypeTime:
|
|
switch v := eval.(type) {
|
|
case []int64:
|
|
row.Time = qbatchTime
|
|
uint64s := make([]uint64, len(v))
|
|
for i := range v {
|
|
if v[i] < 0 {
|
|
return nil, sql3.NewErrInternalf("converting negative slice value to uint64: %d", v[i])
|
|
}
|
|
uint64s[i] = uint64(v[i])
|
|
}
|
|
row.Values[posVals[idx]] = uint64s
|
|
|
|
case []interface{}:
|
|
// it's a tuple, check length
|
|
if len(v) != 2 {
|
|
return nil, sql3.NewErrUnexpectedTimeQuantumTupleLength(0, 0, columnName, rowNumber+1, v, len(v))
|
|
}
|
|
|
|
tupleType, ok := iv.Type().(*parser.DataTypeTuple)
|
|
if !ok {
|
|
return nil, sql3.NewErrInternalf("unexpected tuple type '%T'", v[0])
|
|
}
|
|
|
|
// first member must be a timestamp or coercable as one
|
|
cval, err := coerceValue(tupleType.Members[0], parser.NewDataTypeTimestamp(), v[0], parser.Pos{Line: 0, Column: 0})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
tval, ok := cval.(time.Time)
|
|
if !ok {
|
|
return nil, sql3.NewErrInternalf("unexpected tuple time value type '%T'", v[0])
|
|
}
|
|
var qrowTime fbbatch.QuantizedTime
|
|
qrowTime.Set(tval)
|
|
row.Time = qrowTime
|
|
|
|
// second member must be a set of the correct type
|
|
targetCol := i.targetColumns[idx]
|
|
|
|
switch targetCol.Type().(type) {
|
|
case *parser.DataTypeStringSetQuantum:
|
|
sval, ok := v[1].([]string)
|
|
if !ok {
|
|
return nil, sql3.NewErrInternalf("string set type expected '%T'", v[1])
|
|
}
|
|
row.Values[posVals[idx]] = sval
|
|
|
|
case *parser.DataTypeIDSetQuantum:
|
|
sval, ok := v[1].([]int64)
|
|
if !ok {
|
|
return nil, sql3.NewErrInternalf("id set type expected '%T'", v[1])
|
|
}
|
|
uint64s := make([]uint64, len(sval))
|
|
for i := range sval {
|
|
if sval[i] < 0 {
|
|
return nil, sql3.NewErrInternalf("converting negative slice value to uint64: %d", sval[i])
|
|
}
|
|
uint64s[i] = uint64(sval[i])
|
|
}
|
|
row.Values[posVals[idx]] = uint64s
|
|
|
|
default:
|
|
return nil, sql3.NewErrInternalf("unexpected set type '%T'", targetCol.Type())
|
|
}
|
|
default:
|
|
row.Values[posVals[idx]] = eval
|
|
}
|
|
|
|
case pilosa.FieldTypeInt:
|
|
if eval != nil {
|
|
v, ok := eval.(int64)
|
|
if !ok {
|
|
return nil, sql3.NewErrInternalf("unexpected type %v", eval)
|
|
}
|
|
// check the min and max constraints here
|
|
if v < opts.Min.ToInt64(0) || v > opts.Max.ToInt64(0) {
|
|
return nil, sql3.NewErrInsertValueOutOfRange(0, 0, columnName, rowNumber+1, v)
|
|
}
|
|
}
|
|
row.Values[posVals[idx]] = eval
|
|
|
|
case pilosa.FieldTypeDecimal:
|
|
if eval != nil {
|
|
v, ok := eval.(pql.Decimal)
|
|
if !ok {
|
|
return nil, sql3.NewErrInternalf("unexpected type %v", eval)
|
|
}
|
|
// check the min and max constraints here
|
|
if v.LessThan(opts.Min) || v.GreaterThan(opts.Max) {
|
|
return nil, sql3.NewErrInsertValueOutOfRange(0, 0, columnName, rowNumber+1, v)
|
|
}
|
|
}
|
|
row.Values[posVals[idx]] = eval
|
|
|
|
case pilosa.FieldTypeTimestamp:
|
|
switch v := eval.(type) {
|
|
|
|
// time.Time is used for date literals generated in the parser.
|
|
// For example, if using `current_time`, the type received here
|
|
// will be a time.Time.
|
|
case time.Time:
|
|
// Convert Base, which is the epoch for Timestamp fields, to
|
|
// a time.Time value.
|
|
unit := fbbatch.TimeUnit(opts.TimeUnit)
|
|
epoch, err := fbbatch.Int64ToTimestamp(unit, time.Unix(0, 0), opts.Base)
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "converting base to epoch: %d", opts.Base)
|
|
}
|
|
|
|
i64, err := fbbatch.TimestampToInt64(unit, epoch, v)
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "converting timestamp to int64: %s", v)
|
|
}
|
|
row.Values[posVals[idx]] = i64
|
|
|
|
// string is the normal case for dates; used when the date is
|
|
// provided as a string in the INSERT INTO statement.
|
|
case string:
|
|
ts, err := parser.ConvertStringToTimestamp(v)
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "parsing timestamp: %s", v)
|
|
}
|
|
|
|
// Convert Base, which is the epoch for Timestamp fields, to
|
|
// a time.Time value.
|
|
unit := fbbatch.TimeUnit(opts.TimeUnit)
|
|
epoch, err := fbbatch.Int64ToTimestamp(unit, time.Unix(0, 0), opts.Base)
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "converting base to epoch: %d", opts.Base)
|
|
}
|
|
|
|
i64, err := fbbatch.TimestampToInt64(unit, epoch, ts)
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "converting timestamp to int64: %s", v)
|
|
}
|
|
row.Values[posVals[idx]] = i64
|
|
// integers passed as input for Timestamp fields will be treated as time represented in number of seconds since epoch defined for the field
|
|
// for timestamp fields created using SQL epoch will be defaulted to unix epoch
|
|
case int64:
|
|
// Convert the input seconds to target timeunit defined for the Timestamp field
|
|
// Add Base, which is the base epoch for Timestamp fields, to the input before saving
|
|
unit := fbbatch.TimeUnit(opts.TimeUnit)
|
|
i64 := opts.Base
|
|
switch unit {
|
|
case fbbatch.TimeUnitSeconds:
|
|
i64 = i64 + v
|
|
case fbbatch.TimeUnitMilliseconds:
|
|
i64 = i64 + (v * 1000)
|
|
case fbbatch.TimeUnitMicroseconds, fbbatch.TimeUnitUSeconds:
|
|
i64 = i64 + (v * 1000000)
|
|
case fbbatch.TimeUnitNanoseconds:
|
|
i64 = i64 + (v * 1000000000)
|
|
default:
|
|
return nil, errors.Wrapf(err, "unknown time unit: %s", unit)
|
|
}
|
|
row.Values[posVals[idx]] = i64
|
|
// nil is to support `null` values.
|
|
case nil:
|
|
row.Values[posVals[idx]] = eval
|
|
|
|
default:
|
|
return nil, sql3.NewErrInternalf("unsupported timestamp type: %T", eval)
|
|
}
|
|
|
|
default:
|
|
row.Values[posVals[idx]] = eval
|
|
}
|
|
}
|
|
|
|
if err := batch.Add(row); err != nil {
|
|
// Breaking here on ErrBatchNowFull is only valid because we are
|
|
// explicity setting the batch size to the number of tuples in the
|
|
// INSERT INTO statement. Which means we're only handling a single
|
|
// batch. If this evolves to handle multiple batches, this will need
|
|
// to instead call batch.Import() and continue looping over tuples.
|
|
// We may also need to handle ErrBatchNowStale.
|
|
if err == fbbatch.ErrBatchNowFull {
|
|
break
|
|
}
|
|
return nil, errors.Wrap(err, "adding record")
|
|
}
|
|
}
|
|
|
|
count := batch.Len()
|
|
|
|
if err := batch.Import(); err != nil {
|
|
return nil, errors.Wrap(err, "importing batch")
|
|
}
|
|
|
|
// update the counter for inserts
|
|
pilosa.PerfCounterSQLInsertsSec.Add(int64(count))
|
|
|
|
return nil, types.ErrNoMoreRows
|
|
}
|