// handleCellValue returns a Row if the cell value includes a commit, otherwise nil.
func (cr *chunkReader) handleCellValue(cc *btspb.ReadRowsResponse_CellChunk) Row {
	if cc.ValueSize > 0 {
		// ValueSize is specified so expect a split value of ValueSize bytes
		if cr.curVal == nil {
			cr.curVal = make([]byte, 0, cc.ValueSize)
		}
		cr.curVal = append(cr.curVal, cc.Value...)
		cr.state = cellInProgress
	} else {
		// This cell is either the complete value or the last chunk of a split
		if cr.curVal == nil {
			cr.curVal = cc.Value
		} else {
			cr.curVal = append(cr.curVal, cc.Value...)
		}
		cr.finishCell()

		if cc.GetCommitRow() {
			return cr.commitRow()
		} else {
			cr.state = rowInProgress
		}
	}

	return nil
}

func (cr *chunkReader) validateCellInProgress(cc *btspb.ReadRowsResponse_CellChunk) error {
	if err := cr.validateRowStatus(cc); err != nil {
		return err
	}
	if cr.curVal == nil {
		return fmt.Errorf("no cached cell while CELL_IN_PROGRESS %v", cc)
	}
	if cc.GetResetRow() == false && cr.isAnyKeyPresent(cc) {
		return fmt.Errorf("cell key components found while CELL_IN_PROGRESS %v", cc)
	}
	return nil
}

func (cr *chunkReader) validateNewRow(cc *btspb.ReadRowsResponse_CellChunk) error {
	if cc.GetResetRow() {
		return fmt.Errorf("reset_row not allowed between rows")
	}
	if cc.RowKey == nil || cc.FamilyName == nil || cc.Qualifier == nil {
		return fmt.Errorf("missing key field for new row %v", cc)
	}
	if cr.lastKey != "" && cr.lastKey >= string(cc.RowKey) {
		return fmt.Errorf("out of order row key: %q, %q", cr.lastKey, string(cc.RowKey))
	}
	return nil
}

// Validate a RowStatus, commit or reset, if present.
func (cr *chunkReader) validateRowStatus(cc *btspb.ReadRowsResponse_CellChunk) error {
	// Resets can't be specified with any other part of a cell
	if cc.GetResetRow() && (cr.isAnyKeyPresent(cc) ||
		cc.Value != nil ||
		cc.ValueSize != 0 ||
		cc.Labels != nil) {
		return fmt.Errorf("reset must not be specified with other fields %v", cc)
	}
	if cc.GetCommitRow() && cc.ValueSize > 0 {
		return fmt.Errorf("commit row found in between chunks in a cell")
	}
	return nil
}

// Process takes a cell chunk and returns a new Row if the given chunk
// completes a Row, or nil otherwise.
func (cr *chunkReader) Process(cc *btspb.ReadRowsResponse_CellChunk) (Row, error) {
	var row Row
	switch cr.state {
	case newRow:
		if err := cr.validateNewRow(cc); err != nil {
			return nil, err
		}

		cr.curRow = make(Row)
		cr.curKey = cc.RowKey
		cr.curFam = cc.FamilyName.Value
		cr.curQual = cc.Qualifier.Value
		cr.curTS = cc.TimestampMicros
		row = cr.handleCellValue(cc)

	case rowInProgress:
		if err := cr.validateRowInProgress(cc); err != nil {
			return nil, err
		}

		if cc.GetResetRow() {
			cr.resetToNewRow()
			return nil, nil
		}

		if cc.FamilyName != nil {
			cr.curFam = cc.FamilyName.Value
		}
		if cc.Qualifier != nil {
			cr.curQual = cc.Qualifier.Value
		}
		cr.curTS = cc.TimestampMicros
		row = cr.handleCellValue(cc)

	case cellInProgress:
		if err := cr.validateCellInProgress(cc); err != nil {
			return nil, err
		}
		if cc.GetResetRow() {
			cr.resetToNewRow()
			return nil, nil
		}
		row = cr.handleCellValue(cc)
	}

	return row, nil
}