list_utils -> list

backend/src/apiserver/list/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//backend/api:go_default_library",
+        "//backend/src/apiserver/common:go_default_library",
         "//backend/src/apiserver/filter:go_default_library",
         "//backend/src/common/util:go_default_library",
         "@com_github_masterminds_squirrel//:go_default_library",
@@ -19,10 +20,12 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//backend/api:go_default_library",
+        "//backend/src/apiserver/common:go_default_library",
         "//backend/src/apiserver/filter:go_default_library",
         "//backend/src/common/util:go_default_library",
         "@com_github_google_go_cmp//cmp:go_default_library",
         "@com_github_google_go_cmp//cmp/cmpopts:go_default_library",
         "@com_github_masterminds_squirrel//:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
     ],
 )

backend/src/apiserver/list/list.go

@@ -18,6 +18,7 @@
 package list
 
 import (
+	"database/sql"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
@@ -26,6 +27,7 @@ import (
 
 	sq "github.com/Masterminds/squirrel"
 	api "github.com/kubeflow/pipelines/backend/api/go_client"
+	"github.com/kubeflow/pipelines/backend/src/apiserver/common"
 	"github.com/kubeflow/pipelines/backend/src/apiserver/filter"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 )
@@ -200,6 +202,42 @@ func (o *Options) AddFilterToSelect(sqlBuilder sq.SelectBuilder) sq.SelectBuilde
 	return sqlBuilder
 }
 
+// FilterOnResourceReference filters the given resource's table by rows from the ResourceReferences
+// table that match an optional given filter, and returns the rebuilt SelectBuilder
+func FilterOnResourceReference(tableName string, resourceType common.ResourceType, selectCount bool,
+	filterContext *common.FilterContext) (sq.SelectBuilder, error) {
+	selectBuilder := sq.Select("*")
+	if selectCount {
+		selectBuilder = sq.Select("count(*)")
+	}
+	selectBuilder = selectBuilder.From(tableName)
+	if filterContext.ReferenceKey != nil {
+		resourceReferenceFilter, args, err := sq.Select("ResourceUUID").
+			From("resource_references as rf").
+			Where(sq.And{
+				sq.Eq{"rf.ResourceType": resourceType},
+				sq.Eq{"rf.ReferenceUUID": filterContext.ID},
+				sq.Eq{"rf.ReferenceType": filterContext.Type}}).ToSql()
+		if err != nil {
+			return selectBuilder, util.NewInternalServerError(
+				err, "Failed to create subquery to filter by resource reference: %v", err.Error())
+		}
+		return selectBuilder.Where(fmt.Sprintf("UUID in (%s)", resourceReferenceFilter), args...), nil
+	}
+	return selectBuilder, nil
+}
+
+// Scans the one given row into a number, and returns the number
+func ScanRowToTotalSize(rows *sql.Rows) (int, error) {
+	var total_size int
+	rows.Next()
+	err := rows.Scan(&total_size)
+	if err != nil {
+		return 0, util.NewInternalServerError(err, "Failed to scan row total_size")
+	}
+	return total_size, nil
+}
+
 // Listable is an interface that should be implemented by any resource/model
 // that wants to support listing queries.
 type Listable interface {

backend/src/apiserver/list/list_test.go

@@ -4,8 +4,10 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/kubeflow/pipelines/backend/src/apiserver/common"
 	"github.com/kubeflow/pipelines/backend/src/apiserver/filter"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
+  "github.com/stretchr/testify/assert"
 
 	sq "github.com/Masterminds/squirrel"
 	"github.com/google/go-cmp/cmp"
@@ -605,3 +607,70 @@ func TestMatches(t *testing.T) {
 		}
 	}
 }
+
+func TestFilterOnResourceReference(t *testing.T) {
+
+	type testIn struct {
+		table        string
+		resourceType common.ResourceType
+		count        bool
+		filter       *common.FilterContext
+	}
+	tests := []struct {
+		in      *testIn
+		wantSql string
+		wantErr error
+	}{
+		{
+			in: &testIn{
+				table:        "testTable",
+				resourceType: common.Run,
+				count:        false,
+				filter:       &common.FilterContext{},
+			},
+			wantSql: "SELECT * FROM testTable",
+			wantErr: nil,
+		},
+		{
+			in: &testIn{
+				table:        "testTable",
+				resourceType: common.Run,
+				count:        true,
+				filter:       &common.FilterContext{},
+			},
+			wantSql: "SELECT count(*) FROM testTable",
+			wantErr: nil,
+		},
+		{
+			in: &testIn{
+				table:        "testTable",
+				resourceType: common.Run,
+				count:        false,
+				filter:       &common.FilterContext{ReferenceKey: &common.ReferenceKey{Type: common.Run}},
+			},
+			wantSql: "SELECT * FROM testTable WHERE UUID in (SELECT ResourceUUID FROM resource_references as rf WHERE (rf.ResourceType = ? AND rf.ReferenceUUID = ? AND rf.ReferenceType = ?))",
+			wantErr: nil,
+		},
+		{
+			in: &testIn{
+				table:        "testTable",
+				resourceType: common.Run,
+				count:        true,
+				filter:       &common.FilterContext{ReferenceKey: &common.ReferenceKey{Type: common.Run}},
+			},
+			wantSql: "SELECT count(*) FROM testTable WHERE UUID in (SELECT ResourceUUID FROM resource_references as rf WHERE (rf.ResourceType = ? AND rf.ReferenceUUID = ? AND rf.ReferenceType = ?))",
+			wantErr: nil,
+		},
+	}
+
+	for _, test := range tests {
+		sqlBuilder, gotErr := FilterOnResourceReference(test.in.table, test.in.resourceType, test.in.count, test.in.filter)
+		gotSql, _, err := sqlBuilder.ToSql()
+		assert.Nil(t, err)
+
+		if gotSql != test.wantSql || gotErr != test.wantErr {
+			t.Errorf("FilterOnResourceReference(%+v) =\nGot: %q, %v\nWant: %q, %v",
+				test.in, gotSql, gotErr, test.wantSql, test.wantErr)
+		}
+	}
+}

backend/src/apiserver/storage/BUILD.bazel

@@ -8,7 +8,6 @@ go_library(
         "db_status_store.go",
         "experiment_store.go",
         "job_store.go",
-        "list_util.go",
         "minio_client.go",
         "minio_client_fake.go",
         "object_store.go",
@@ -47,7 +46,6 @@ go_test(
         "db_test.go",
         "experiment_store_test.go",
         "job_store_test.go",
-        "list_util_test.go",
         "object_store_test.go",
         "pipeline_store_test.go",
         "resource_reference_store_test.go",

backend/src/apiserver/storage/experiment_store.go

@@ -73,7 +73,7 @@ func (s *ExperimentStore) ListExperiments(opts *list.Options) ([]*model.Experime
 		return errorF(err)
 	}
 	defer sizeRow.Close()
-	total_size, err := ScanRowToTotalSize(sizeRow)
+	total_size, err := list.ScanRowToTotalSize(sizeRow)
 	if err != nil {
 		tx.Rollback()
 		return errorF(err)

backend/src/apiserver/storage/job_store.go

@@ -84,7 +84,7 @@ func (s *JobStore) ListJobs(
 		return errorF(err)
 	}
 	defer sizeRow.Close()
-	total_size, err := ScanRowToTotalSize(sizeRow)
+	total_size, err := list.ScanRowToTotalSize(sizeRow)
 	if err != nil {
 		tx.Rollback()
 		return errorF(err)
@@ -106,7 +106,7 @@ func (s *JobStore) ListJobs(
 
 func (s *JobStore) buildSelectJobsQuery(selectCount bool, opts *list.Options,
 	filterContext *common.FilterContext) (string, []interface{}, error) {
-	filteredSelectBuilder, err := FilterOnResourceReference("jobs", common.Job, selectCount, filterContext)
+	filteredSelectBuilder, err := list.FilterOnResourceReference("jobs", common.Job, selectCount, filterContext)
 	if err != nil {
 		return "", nil, util.NewInternalServerError(err, "Failed to list jobs: %v", err)
 	}

backend/src/apiserver/storage/pipeline_store.go

@@ -78,24 +78,24 @@ func (s *PipelineStore) ListPipelines(opts *list.Options) ([]*model.Pipeline, in
 		tx.Rollback()
 		return errorF(err)
 	}
-	defer rows.Close()
 	pipelines, err := s.scanRows(rows)
 	if err != nil {
 		tx.Rollback()
 		return errorF(err)
 	}
+	rows.Close()
 
 	sizeRow, err := tx.Query(sizeSql, sizeArgs...)
 	if err != nil {
 		tx.Rollback()
 		return errorF(err)
 	}
-	defer sizeRow.Close()
-	total_size, err := ScanRowToTotalSize(sizeRow)
+	total_size, err := list.ScanRowToTotalSize(sizeRow)
 	if err != nil {
 		tx.Rollback()
 		return errorF(err)
 	}
+	sizeRow.Close()
 
 	err = tx.Commit()
 	if err != nil {

backend/src/apiserver/storage/run_store.go

@@ -104,7 +104,7 @@ func (s *RunStore) ListRuns(
 		return errorF(err)
 	}
 	defer sizeRow.Close()
-	total_size, err := ScanRowToTotalSize(sizeRow)
+	total_size, err := list.ScanRowToTotalSize(sizeRow)
 	if err != nil {
 		tx.Rollback()
 		return errorF(err)
@@ -132,7 +132,7 @@ func (s *RunStore) ListRuns(
 
 func (s *RunStore) buildSelectRunsQuery(selectCount bool, opts *list.Options,
 	filterContext *common.FilterContext) (string, []interface{}, error) {
-	filteredSelectBuilder, err := FilterOnResourceReference("run_details", common.Run, selectCount, filterContext)
+	filteredSelectBuilder, err := list.FilterOnResourceReference("run_details", common.Run, selectCount, filterContext)
 	if err != nil {
 		return "", nil, util.NewInternalServerError(err, "Failed to list runs: %v", err)
 	}