// Import bulk imports data into a particular index,field,shard.
Import bulk imports data into a particular index,field,shard.
// Import bulk imports data into a particular index field shard .
func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOption) error {
	span, _ := tracing.StartSpanFromContext(ctx, "API.Import")
	defer span.Finish()
	if err := api.validate(apiImport); 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 field.keys() {
			if len(req.RowIDs) != 0 {
				return errors.New("row ids cannot be used because field uses string keys")
			}
			if req.RowIDs, err = api.holder.translateFile.TranslateRowsToUint64(index.Name(), field.Name(), req.RowKeys); err != nil {
				return errors.Wrap(err, "translating rows")
			}
		}
		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")
			}
		}
		if index.Keys() || field.keys() {
			m := make(map[uint64][]Bit)
			for i, colID := range req.ColumnIDs {
				shard := colID / ShardWidth
				if _, ok := m[shard]; !ok {
					m[shard] = make([]Bit, 0)
				}
				m[shard] = append(m[shard], Bit{
					RowID:     req.RowIDs[i],
					ColumnID:  colID,
					Timestamp: req.Timestamps[i],
				})
			}
			opts = append(opts, OptImportOptionsIgnoreKeyCheck(true))
			var eg errgroup.Group
			for shard, bits := range m {
				shard := shard
				bits := bits
				eg.Go(func() error {
					return api.server.defaultClient.Import(ctx, req.Index, req.Field, shard, bits, opts...)
				})
			}
			return eg.Wait()
		}
	}
	if err := api.validateShardOwnership(req.Index, req.Shard); err != nil {
		return errors.Wrap(err, "validating shard ownership")
	}
	timestamps := make([]*time.Time, len(req.Timestamps))
	for i, ts := range req.Timestamps {
		if ts == 0 {
			continue
		}
		t := time.Unix(0, ts).UTC()
		timestamps[i] = &t
	}
	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.Import(req.RowIDs, req.ColumnIDs, timestamps, opts...)
	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 ) Import ( ctx context . Context , req * ImportRequest , opts ... ImportOption ) error { span , _ := tracing . StartSpanFromContext ( ctx , " " ) defer span . Finish ( ) if err := api . validate ( apiImport ) ; 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 field . keys ( ) { if len ( req . RowIDs ) != 0 { return errors . New ( " " ) } if req . RowIDs , err = api . holder . translateFile . TranslateRowsToUint64 ( index . Name ( ) , field . Name ( ) , req . RowKeys ) ; err != nil { return errors . Wrap ( err , " " ) } } 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 , " " ) } } if index . Keys ( ) || field . keys ( ) { m := make ( map [ uint64 ] [ ] Bit ) for i , colID := range req . ColumnIDs { shard := colID / ShardWidth if _ , ok := m [ shard ] ; ! ok { m [ shard ] = make ( [ ] Bit , 0 ) } m [ shard ] = append ( m [ shard ] , Bit { RowID : req . RowIDs [ i ] , ColumnID : colID , Timestamp : req . Timestamps [ i ] , } ) } opts = append ( opts , OptImportOptionsIgnoreKeyCheck ( true ) ) var eg errgroup . Group for shard , bits := range m { shard := shard bits := bits eg . Go ( func ( ) error { return api . server . defaultClient . Import ( ctx , req . Index , req . Field , shard , bits , opts ... ) } ) } return eg . Wait ( ) } } if err := api . validateShardOwnership ( req . Index , req . Shard ) ; err != nil { return errors . Wrap ( err , " " ) } timestamps := make ( [ ] * time . Time , len ( req . Timestamps ) ) for i , ts := range req . Timestamps { if ts == 0 { continue } t := time . Unix ( 0 , ts ) . UTC ( ) timestamps [ i ] = & t } 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 . Import ( req . RowIDs , req . ColumnIDs , timestamps , opts ... ) if err != nil { api . server . logger . Printf ( " " , req . Index , req . Field , req . Shard , len ( req . ColumnIDs ) , err ) } return errors . Wrap ( err , " " ) }
Import bulk imports data into a particular index,field,shard.
Import bulk imports data into a particular index field shard .
// Set up import options. // Unless explicitly ignoring key validation (meaning keys have been // translated to ids in a previous step at the coordinator node), then // check to see if keys need translation. // Translate row keys. // 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. // Convert timestamps to time.Time. // Import columnIDs into existence field. // Import into fragment.