func (tb *table) AddRows(rows []rawRow) error {
	if len(rows) == 0 {
		return nil
	}

	// Verify whether all the rows may be added to a single partition.
	ptwsX := getPartitionWrappers()
	defer putPartitionWrappers(ptwsX)

	ptwsX.a = tb.GetPartitions(ptwsX.a[:0])
	ptws := ptwsX.a
	for _, ptw := range ptws {
		singlePt := true
		for i := range rows {
			if !ptw.pt.HasTimestamp(rows[i].Timestamp) {
				singlePt = false
				break
			}
		}
		if !singlePt {
			continue
		}

		// Move the partition with the matching rows to the front of tb.ptws,
		// so it will be detected faster next time.
		tb.ptwsLock.Lock()
		for i := range tb.ptws {
			if ptw == tb.ptws[i] {
				tb.ptws[0], tb.ptws[i] = tb.ptws[i], tb.ptws[0]
				break
			}
		}
		tb.ptwsLock.Unlock()

		// Fast path - add all the rows into the ptw.
		ptw.pt.AddRows(rows)
		tb.PutPartitions(ptws)
		return nil
	}

	// Slower path - split rows into per-partition buckets.
	ptBuckets := make(map[*partitionWrapper][]rawRow)
	var missingRows []rawRow
	for i := range rows {
		r := &rows[i]
		ptFound := false
		for _, ptw := range ptws {
			if ptw.pt.HasTimestamp(r.Timestamp) {
				ptBuckets[ptw] = append(ptBuckets[ptw], *r)
				ptFound = true
				break
			}
		}
		if !ptFound {
			missingRows = append(missingRows, *r)
		}
	}

	for ptw, ptRows := range ptBuckets {
		ptw.pt.AddRows(ptRows)
	}
	tb.PutPartitions(ptws)
	if len(missingRows) == 0 {
		return nil
	}

	// The slowest path - there are rows that don't fit any existing partition.
	// Create new partitions for these rows.
	// Do this under tb.ptwsLock.
	now := timestampFromTime(time.Now())
	minTimestamp := now - tb.retentionMilliseconds
	maxTimestamp := now + 2*24*3600*1000 // allow max +2 days from now due to timezones shit :)
	tb.ptwsLock.Lock()
	var errors []error
	for i := range missingRows {
		r := &missingRows[i]

		if r.Timestamp < minTimestamp {
			// Silently skip row with too small timestamp, since it should be deleted anyway.
			continue
		}
		if r.Timestamp > maxTimestamp {
			err := fmt.Errorf("cannot add row %+v with too big timestamp to table %q; the timestamp cannot be bigger than %d (+2 days from now)",
				r, tb.path, maxTimestamp)
			errors = append(errors, err)
			continue
		}

		// Make sure the partition for the r hasn't been added by another goroutines.
		ptFound := false
		for _, ptw := range tb.ptws {
			if ptw.pt.HasTimestamp(r.Timestamp) {
				ptFound = true
				ptw.pt.AddRows(missingRows[i : i+1])
				break
			}
		}
		if ptFound {
			continue
		}

		pt, err := createPartition(r.Timestamp, tb.smallPartitionsPath, tb.bigPartitionsPath, tb.getDeletedMetricIDs)
		if err != nil {
			errors = append(errors, err)
			continue
		}
		pt.AddRows(missingRows[i : i+1])
		tb.addPartitionNolock(pt)
	}
	tb.ptwsLock.Unlock()

	if len(errors) > 0 {
		// Return only the first error, since it has no sense in returning all errors.
		return fmt.Errorf("errors while adding rows to table %q: %s", tb.path, errors[0])
	}
	return nil
}
// AddRows adds the given rows to the table tb.
AddRows adds the given rows to the table tb.
// AddRows adds the given rows to the table tb .
func (tb *table) AddRows(rows []rawRow) error {
	if len(rows) == 0 {
		return nil
	}
	ptwsX := getPartitionWrappers()
	defer putPartitionWrappers(ptwsX)
	ptwsX.a = tb.GetPartitions(ptwsX.a[:0])
	ptws := ptwsX.a
	for _, ptw := range ptws {
		singlePt := true
		for i := range rows {
			if !ptw.pt.HasTimestamp(rows[i].Timestamp) {
				singlePt = false
				break
			}
		}
		if !singlePt {
			continue
		}
		tb.ptwsLock.Lock()
		for i := range tb.ptws {
			if ptw == tb.ptws[i] {
				tb.ptws[0], tb.ptws[i] = tb.ptws[i], tb.ptws[0]
				break
			}
		}
		tb.ptwsLock.Unlock()
		ptw.pt.AddRows(rows)
		tb.PutPartitions(ptws)
		return nil
	}
	ptBuckets := make(map[*partitionWrapper][]rawRow)
	var missingRows []rawRow
	for i := range rows {
		r := &rows[i]
		ptFound := false
		for _, ptw := range ptws {
			if ptw.pt.HasTimestamp(r.Timestamp) {
				ptBuckets[ptw] = append(ptBuckets[ptw], *r)
				ptFound = true
				break
			}
		}
		if !ptFound {
			missingRows = append(missingRows, *r)
		}
	}
	for ptw, ptRows := range ptBuckets {
		ptw.pt.AddRows(ptRows)
	}
	tb.PutPartitions(ptws)
	if len(missingRows) == 0 {
		return nil
	}
	now := timestampFromTime(time.Now())
	minTimestamp := now - tb.retentionMilliseconds
	maxTimestamp := now + 2*24*3600*1000 
	tb.ptwsLock.Lock()
	var errors []error
	for i := range missingRows {
		r := &missingRows[i]
		if r.Timestamp < minTimestamp {
			continue
		}
		if r.Timestamp > maxTimestamp {
			err := fmt.Errorf("cannot add row %+v with too big timestamp to table %q; the timestamp cannot be bigger than %d (+2 days from now)",
				r, tb.path, maxTimestamp)
			errors = append(errors, err)
			continue
		}
		ptFound := false
		for _, ptw := range tb.ptws {
			if ptw.pt.HasTimestamp(r.Timestamp) {
				ptFound = true
				ptw.pt.AddRows(missingRows[i : i+1])
				break
			}
		}
		if ptFound {
			continue
		}
		pt, err := createPartition(r.Timestamp, tb.smallPartitionsPath, tb.bigPartitionsPath, tb.getDeletedMetricIDs)
		if err != nil {
			errors = append(errors, err)
			continue
		}
		pt.AddRows(missingRows[i : i+1])
		tb.addPartitionNolock(pt)
	}
	tb.ptwsLock.Unlock()
	if len(errors) > 0 {
		return fmt.Errorf("errors while adding rows to table %q: %s", tb.path, errors[0])
	}
	return nil
}
func ( tb * table ) AddRows ( rows [ ] rawRow ) error { if len ( rows ) == 0 { return nil } ptwsX := getPartitionWrappers ( ) defer putPartitionWrappers ( ptwsX ) ptwsX . a = tb . GetPartitions ( ptwsX . a [ : 0 ] ) ptws := ptwsX . a for _ , ptw := range ptws { singlePt := true for i := range rows { if ! ptw . pt . HasTimestamp ( rows [ i ] . Timestamp ) { singlePt = false break } } if ! singlePt { continue } tb . ptwsLock . Lock ( ) for i := range tb . ptws { if ptw == tb . ptws [ i ] { tb . ptws [ 0 ] , tb . ptws [ i ] = tb . ptws [ i ] , tb . ptws [ 0 ] break } } tb . ptwsLock . Unlock ( ) ptw . pt . AddRows ( rows ) tb . PutPartitions ( ptws ) return nil } ptBuckets := make ( map [ * partitionWrapper ] [ ] rawRow ) var missingRows [ ] rawRow for i := range rows { r := & rows [ i ] ptFound := false for _ , ptw := range ptws { if ptw . pt . HasTimestamp ( r . Timestamp ) { ptBuckets [ ptw ] = append ( ptBuckets [ ptw ] , * r ) ptFound = true break } } if ! ptFound { missingRows = append ( missingRows , * r ) } } for ptw , ptRows := range ptBuckets { ptw . pt . AddRows ( ptRows ) } tb . PutPartitions ( ptws ) if len ( missingRows ) == 0 { return nil } now := timestampFromTime ( time . Now ( ) ) minTimestamp := now - tb . retentionMilliseconds maxTimestamp := now + 2 * 24 * 3600 * 1000 tb . ptwsLock . Lock ( ) var errors [ ] error for i := range missingRows { r := & missingRows [ i ] if r . Timestamp < minTimestamp { continue } if r . Timestamp > maxTimestamp { err := fmt . Errorf ( " " , r , tb . path , maxTimestamp ) errors = append ( errors , err ) continue } ptFound := false for _ , ptw := range tb . ptws { if ptw . pt . HasTimestamp ( r . Timestamp ) { ptFound = true ptw . pt . AddRows ( missingRows [ i : i + 1 ] ) break } } if ptFound { continue } pt , err := createPartition ( r . Timestamp , tb . smallPartitionsPath , tb . bigPartitionsPath , tb . getDeletedMetricIDs ) if err != nil { errors = append ( errors , err ) continue } pt . AddRows ( missingRows [ i : i + 1 ] ) tb . addPartitionNolock ( pt ) } tb . ptwsLock . Unlock ( ) if len ( errors ) > 0 { return fmt . Errorf ( " " , tb . path , errors [ 0 ] ) } return nil }
AddRows adds the given rows to the table tb.
AddRows adds the given rows to the table tb .
// Verify whether all the rows may be added to a single partition. // Move the partition with the matching rows to the front of tb.ptws, // so it will be detected faster next time. // Fast path - add all the rows into the ptw. // Slower path - split rows into per-partition buckets. // The slowest path - there are rows that don't fit any existing partition. // Create new partitions for these rows. // Do this under tb.ptwsLock. // allow max +2 days from now due to timezones shit :) // Silently skip row with too small timestamp, since it should be deleted anyway. // Make sure the partition for the r hasn't been added by another goroutines. // Return only the first error, since it has no sense in returning all errors.