// ImportValue bulk imports values into a particular field.
ImportValue bulk imports values into a particular field.
// ImportValue bulk imports values into a particular field .
func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts ...ImportOption) error {
	span, _ := tracing.StartSpanFromContext(ctx, "API.ImportValue")
	defer span.Finish()
	if err := api.validate(apiImportValue); err != nil {
		return errors.Wrap(err, "validating api method")
	}
	options, err := setUpImportOptions(opts...)
	if err != nil {
		return errors.Wrap(err, "setting up import options")
	}
	index, field, err := api.indexField(req.Index, req.Field, req.Shard)
	if err != nil {
		return errors.Wrap(err, "getting index and field")
	}
	if !options.IgnoreKeyCheck {
		if index.Keys() {
			if len(req.ColumnIDs) != 0 {
				return errors.New("column ids cannot be used because index uses string keys")
			}
			if req.ColumnIDs, err = api.holder.translateFile.TranslateColumnsToUint64(index.Name(), req.ColumnKeys); err != nil {
				return errors.Wrap(err, "translating columns")
			}
			m := make(map[uint64][]FieldValue)
			for i, colID := range req.ColumnIDs {
				shard := colID / ShardWidth
				if _, ok := m[shard]; !ok {
					m[shard] = make([]FieldValue, 0)
				}
				m[shard] = append(m[shard], FieldValue{
					Value:    req.Values[i],
					ColumnID: colID,
				})
			}
			opts = append(opts, OptImportOptionsIgnoreKeyCheck(true))
			var eg errgroup.Group
			for shard, vals := range m {
				shard := shard
				vals := vals
				eg.Go(func() error {
					return api.server.defaultClient.ImportValue(ctx, req.Index, req.Field, shard, vals, opts...)
				})
			}
			return eg.Wait()
		}
	}
	if err := api.validateShardOwnership(req.Index, req.Shard); err != nil {
		return errors.Wrap(err, "validating shard ownership")
	}
	if !options.Clear {
		if err := importExistenceColumns(index, req.ColumnIDs); err != nil {
			api.server.logger.Printf("import existence error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
			return errors.Wrap(err, "importing existence columns")
		}
	}
	err = field.importValue(req.ColumnIDs, req.Values, options)
	if err != nil {
		api.server.logger.Printf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
	}
	return errors.Wrap(err, "importing")
}
func ( api * API ) ImportValue ( ctx context . Context , req * ImportValueRequest , opts ... ImportOption ) error { span , _ := tracing . StartSpanFromContext ( ctx , " " ) defer span . Finish ( ) if err := api . validate ( apiImportValue ) ; err != nil { return errors . Wrap ( err , " " ) } options , err := setUpImportOptions ( opts ... ) if err != nil { return errors . Wrap ( err , " " ) } index , field , err := api . indexField ( req . Index , req . Field , req . Shard ) if err != nil { return errors . Wrap ( err , " " ) } if ! options . IgnoreKeyCheck { if index . Keys ( ) { if len ( req . ColumnIDs ) != 0 { return errors . New ( " " ) } if req . ColumnIDs , err = api . holder . translateFile . TranslateColumnsToUint64 ( index . Name ( ) , req . ColumnKeys ) ; err != nil { return errors . Wrap ( err , " " ) } m := make ( map [ uint64 ] [ ] FieldValue ) for i , colID := range req . ColumnIDs { shard := colID / ShardWidth if _ , ok := m [ shard ] ; ! ok { m [ shard ] = make ( [ ] FieldValue , 0 ) } m [ shard ] = append ( m [ shard ] , FieldValue { Value : req . Values [ i ] , ColumnID : colID , } ) } opts = append ( opts , OptImportOptionsIgnoreKeyCheck ( true ) ) var eg errgroup . Group for shard , vals := range m { shard := shard vals := vals eg . Go ( func ( ) error { return api . server . defaultClient . ImportValue ( ctx , req . Index , req . Field , shard , vals , opts ... ) } ) } return eg . Wait ( ) } } if err := api . validateShardOwnership ( req . Index , req . Shard ) ; err != nil { return errors . Wrap ( err , " " ) } if ! options . Clear { if err := importExistenceColumns ( index , req . ColumnIDs ) ; err != nil { api . server . logger . Printf ( " " , req . Index , req . Field , req . Shard , len ( req . ColumnIDs ) , err ) return errors . Wrap ( err , " " ) } } err = field . importValue ( req . ColumnIDs , req . Values , options ) if err != nil { api . server . logger . Printf ( " " , req . Index , req . Field , req . Shard , len ( req . ColumnIDs ) , err ) } return errors . Wrap ( err , " " ) }
ImportValue bulk imports values into a particular field.
ImportValue bulk imports values into a particular field .
// Set up import options. // Unless explicitly ignoring key validation (meaning keys have been // translate to ids in a previous step at the coordinator node), then // check to see if keys need translation. // Translate column keys. // For translated data, map the columnIDs to shards. If // this node does not own the shard, forward to the node that does. // Signal to the receiving nodes to ignore checking for key translation. // TODO: if local node owns this shard we don't need to go through the client // Validate shard ownership. // Import columnIDs into existence field. // Import into fragment.