*: don't read blob content if it's not used

Blob contents should not be read unless the blobs.content column is specifically used in the query (either in direct use or in a projection) to save memory. Operations such as aggregations may use a huge amount of memory without this optimisation.

Thanks to the column pushdown we are able to know if the column is in use, so if it is, just flag the iterators to let them know they should read the content.

Signed-off-by: Miguel Molina <[email protected]>

Author: erizocosmico

blobs.go

@@ -6,6 +6,7 @@ import (
 	"io/ioutil"
 
 	"gopkg.in/src-d/go-mysql-server.v0/sql"
+	"gopkg.in/src-d/go-mysql-server.v0/sql/expression"
 
 	"gopkg.in/src-d/go-git.v4/plumbing"
 	"gopkg.in/src-d/go-git.v4/plumbing/object"
@@ -70,7 +71,7 @@ func (r *blobsTable) TransformExpressionsUp(f sql.TransformExprFunc) (sql.Node,
 
 func (r blobsTable) RowIter(ctx *sql.Context) (sql.RowIter, error) {
 	span, ctx := ctx.Span("gitbase.BlobsTable")
-	iter := new(blobIter)
+	iter := &blobIter{readContent: true}
 
 	repoIter, err := NewRowRepoIter(ctx, iter)
 	if err != nil {
@@ -91,23 +92,26 @@ func (blobsTable) HandledFilters(filters []sql.Expression) []sql.Expression {
 
 func (r *blobsTable) WithProjectAndFilters(
 	ctx *sql.Context,
-	_, filters []sql.Expression,
+	columns, filters []sql.Expression,
 ) (sql.RowIter, error) {
 	span, ctx := ctx.Span("gitbase.BlobsTable")
 	iter, err := rowIterWithSelectors(
 		ctx, BlobsSchema, BlobsTableName, filters,
 		[]string{"hash"},
 		func(selectors selectors) (RowRepoIter, error) {
 			if len(selectors["hash"]) == 0 {
-				return new(blobIter), nil
+				return &blobIter{readContent: shouldReadContent(columns)}, nil
 			}
 
 			hashes, err := selectors.textValues("hash")
 			if err != nil {
 				return nil, err
 			}
 
-			return &blobsByHashIter{hashes: hashes}, nil
+			return &blobsByHashIter{
+				hashes:      hashes,
+				readContent: shouldReadContent(columns),
+			}, nil
 		},
 	)
 
@@ -120,7 +124,8 @@ func (r *blobsTable) WithProjectAndFilters(
 }
 
 type blobIter struct {
-	iter *object.BlobIter
+	iter        *object.BlobIter
+	readContent bool
 }
 
 func (i *blobIter) NewIterator(repo *Repository) (RowRepoIter, error) {
@@ -129,7 +134,7 @@ func (i *blobIter) NewIterator(repo *Repository) (RowRepoIter, error) {
 		return nil, err
 	}
 
-	return &blobIter{iter: iter}, nil
+	return &blobIter{iter: iter, readContent: i.readContent}, nil
 }
 
 func (i *blobIter) Next() (sql.Row, error) {
@@ -138,7 +143,7 @@ func (i *blobIter) Next() (sql.Row, error) {
 		return nil, err
 	}
 
-	return blobToRow(o)
+	return blobToRow(o, i.readContent)
 }
 
 func (i *blobIter) Close() error {
@@ -150,13 +155,14 @@ func (i *blobIter) Close() error {
 }
 
 type blobsByHashIter struct {
-	repo   *Repository
-	pos    int
-	hashes []string
+	repo        *Repository
+	pos         int
+	hashes      []string
+	readContent bool
 }
 
 func (i *blobsByHashIter) NewIterator(repo *Repository) (RowRepoIter, error) {
-	return &blobsByHashIter{repo, 0, i.hashes}, nil
+	return &blobsByHashIter{repo, 0, i.hashes, i.readContent}, nil
 }
 
 func (i *blobsByHashIter) Next() (sql.Row, error) {
@@ -176,26 +182,26 @@ func (i *blobsByHashIter) Next() (sql.Row, error) {
 			return nil, err
 		}
 
-		return blobToRow(blob)
+		return blobToRow(blob, i.readContent)
 	}
 }
 
 func (i *blobsByHashIter) Close() error {
 	return nil
 }
 
-func blobToRow(c *object.Blob) (sql.Row, error) {
+func blobToRow(c *object.Blob, readContent bool) (sql.Row, error) {
 	var content []byte
 	var isAllowed = blobsAllowBinary
-	if !isAllowed {
+	if !isAllowed && readContent {
 		ok, err := isBinary(c)
 		if err != nil {
 			return nil, err
 		}
 		isAllowed = !ok
 	}
 
-	if c.Size <= int64(blobsMaxSize) && isAllowed {
+	if c.Size <= int64(blobsMaxSize) && isAllowed && readContent {
 		r, err := c.Reader()
 		if err != nil {
 			return nil, err
@@ -248,3 +254,19 @@ func isBinary(blob *object.Blob) (bool, error) {
 		}
 	}
 }
+
+func shouldReadContent(columns []sql.Expression) bool {
+	for _, e := range columns {
+		var found bool
+		expression.Inspect(e, func(e sql.Expression) bool {
+			gf, ok := e.(*expression.GetField)
+			found = ok && gf.Table() == BlobsTableName && gf.Name() == "content"
+			return !found
+		})
+
+		if found {
+			return true
+		}
+	}
+	return false
+}

integration_test.go

@@ -229,10 +229,10 @@ func TestSquashCorrectness(t *testing.T) {
 		`SELECT * FROM tree_entries te INNER JOIN blobs b ON te.entry_hash = b.hash`,
 
 		`SELECT * FROM repositories r
-		INNER JOIN refs re 
+		INNER JOIN refs re
 			ON r.id = re.repository_id
-		INNER JOIN commits c 
-			ON re.hash = c.hash 
+		INNER JOIN commits c
+			ON re.hash = c.hash
 		WHERE re.name = 'HEAD'`,
 
 		`SELECT * FROM commits c

internal/rule/squashjoins.go

@@ -312,6 +312,14 @@ func buildSquashedTable(
 				return nil, errInvalidIteratorChain.New("tree_entries", iter)
 			}
 		case gitbase.BlobsTableName:
+			var readContent bool
+			for _, e := range columns {
+				if containsField(e, gitbase.BlobsTableName, "content") {
+					readContent = true
+					break
+				}
+			}
+
 			switch it := iter.(type) {
 			case gitbase.RefsIter:
 				var f sql.Expression
@@ -328,6 +336,7 @@ func buildSquashedTable(
 				iter = gitbase.NewCommitBlobsIter(
 					gitbase.NewRefHEADCommitsIter(it, nil, true),
 					f,
+					readContent,
 				)
 			case gitbase.CommitsIter:
 				var f sql.Expression
@@ -341,13 +350,10 @@ func buildSquashedTable(
 					return nil, err
 				}
 
-				iter = gitbase.NewTreeEntryBlobsIter(
-					gitbase.NewCommitMainTreeEntriesIter(
-						it,
-						nil,
-						true,
-					),
+				iter = gitbase.NewCommitBlobsIter(
+					it,
 					f,
+					readContent,
 				)
 			case gitbase.TreeEntriesIter:
 				var f sql.Expression
@@ -361,7 +367,7 @@ func buildSquashedTable(
 					return nil, err
 				}
 
-				iter = gitbase.NewTreeEntryBlobsIter(it, f)
+				iter = gitbase.NewTreeEntryBlobsIter(it, f, readContent)
 			default:
 				return nil, errInvalidIteratorChain.New("blobs", iter)
 			}
@@ -936,6 +942,19 @@ func isNum(n int64) validator {
 	}
 }
 
+func containsField(e sql.Expression, table, name string) bool {
+	var found bool
+	expression.Inspect(e, func(e sql.Expression) bool {
+		gf, ok := e.(*expression.GetField)
+		if ok && gf.Table() == table && gf.Name() == name {
+			found = true
+			return false
+		}
+		return true
+	})
+	return found
+}
+
 func fixFieldIndexes(e sql.Expression, schema sql.Schema) (sql.Expression, error) {
 	if e == nil {
 		return nil, nil