LogDog: Enable keys-only BigTable queries.

server/logdog/storage/bigtable/bigtable.go

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package bigtable
 
 import (
         "fmt"
         "time"
 
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/grpcutil"
         "github.com/luci/luci-go/server/logdog/storage"
         "golang.org/x/net/context"
         "google.golang.org/cloud/bigtable"
 )
 
 const (
         logColumnFamily = "log"
-        logColumn       = "data"
+
+        // The data column stores raw low row data (RecordIO blob).
+        logColumn  = "data"
+        logColName = logColumnFamily + ":" + logColumn
 )
 
 // Limits taken from here:
 // https://cloud.google.com/bigtable/docs/schema-design
 const (
         // bigTableRowMaxBytes is the maximum number of bytes that a single BigTable
         // row may hold.
         bigTableRowMaxBytes = 1024 * 1024 * 10 // 10MB
 )
 
@@ skipping 49 matching lines @@
         }
         return nil
 }
 
 func (t *btTableProd) getLogData(c context.Context, rk *rowKey, limit int, keysOnly bool, cb btGetCallback) error {
         // Construct read options based on Get request.
         ropts := []bigtable.ReadOption{
                 bigtable.RowFilter(bigtable.FamilyFilter(logColumnFamily)),
                 bigtable.RowFilter(bigtable.ColumnFilter(logColumn)),
                 nil,
-                nil,
         }[:2]
         if keysOnly {
                 ropts = append(ropts, bigtable.RowFilter(bigtable.StripValueFilter()))
         }
         if limit > 0 {
                 ropts = append(ropts, bigtable.LimitRows(int64(limit)))
         }
 
         // This will limit the range to the immediate row key ("ASDF~INDEX") to
         // immediately after the row key ("ASDF~~"). See rowKey for more information.
         rng := bigtable.NewRange(rk.encode(), rk.pathPrefixUpperBound())
 
-        innerErr := error(nil)
+        var innerErr error
         err := t.logTable.ReadRows(c, rng, func(row bigtable.Row) bool {
-                data := []byte(nil)
-                if !keysOnly {
-                        err := error(nil)
-                        data, err = getLogData(row)
-                        if err != nil {
-                                innerErr = storage.ErrBadData
-                                return false
-                        }
+                data, err := getLogRowData(row)
+                if err != nil {
+                        innerErr = storage.ErrBadData
+                        return false
                 }
 
                 drk, err := decodeRowKey(row.Key())
                 if err != nil {
                         innerErr = err
                         return false
                 }
 
                 if err := cb(drk, data); err != nil {
                         innerErr = err
@@ skipping 28 matching lines @@
 }
 
 // isTransient tests if a BigTable SDK error is transient.
 //
 // Since the BigTable API doesn't give us this information, we will identify
 // transient errors by parsing their error string :(
 func isTransient(err error) bool {
         return (err != errStop) && grpcutil.IsTransient(err)
 }
 
-// getLogData loads the logColumn column from the logColumnFamily column family
-// and returns its []byte contents.
+// getLogRowData loads the []byte contents of the supplied log row.
 //
 // If the row doesn't exist, storage.ErrDoesNotExist will be returned.
-func getLogData(row bigtable.Row) ([]byte, error) {
-        ri := getReadItem(row, logColumnFamily, logColumn)
-        if ri == nil {
-                return nil, storage.ErrDoesNotExist
+func getLogRowData(row bigtable.Row) (data []byte, err error) {
+        items, ok := row[logColumnFamily]
+        if !ok {
+                err = storage.ErrDoesNotExist
+                return
         }
-        return ri.Value, nil
+
+        for _, item := range items {
+                switch item.Column {
+                case logColName:
+                        data = item.Value
+                        return
+                }
+        }
+
+        // If no fields could be extracted, the rows does not exist.
+        err = storage.ErrDoesNotExist
+        return
 }
 
 // getReadItem retrieves a specific RowItem from the supplied Row.
 func getReadItem(row bigtable.Row, family, column string) *bigtable.ReadItem {
         // Get the row for our family.
-        items, ok := row[family]
+        items, ok := row[logColumnFamily]
         if !ok {
                 return nil
         }
 
         // Get the specific ReadItem for our column
         colName := fmt.Sprintf("%s:%s", family, column)
         for _, item := range items {
                 if item.Column == colName {
                         return &item
                 }
         }
         return nil
 }