Repository: oom-ai/oomstore
Pull request #1176 (2022-02-10 09:29:42)

refactor: clean up code

wfxr: suggested commit message: `refactor: clean up code`

Patch:
diff --git a/internal/database/dbutil/schema.go b/internal/database/dbutil/schema.go
index ce9124e9..8ff1bebf 100644
--- a/internal/database/dbutil/schema.go
+++ b/internal/database/dbutil/schema.go
@@ -23,18 +23,16 @@ func (c ColumnList) Names() []string {
 	return names
 }
 
-type BuildTableSchemaParams struct {
-	TableName    string
-	EntityName   string
-	HasUnixMilli bool
-	Features     types.FeatureList
-	PrimaryKeys  []string
-	Backend      types.BackendType
-}
-
-func BuildTableSchema(params BuildTableSchemaParams) string {
-	columns := parseColumns(params.EntityName, params.HasUnixMilli, params.Features, params.Backend)
-	return createTableDDL(params.TableName, columns, params.PrimaryKeys, params.Backend)
+func BuildTableSchema(
+	tableName string,
+	entityName string,
+	hasUnixMilli bool,
+	features types.FeatureList,
+	pkFields []string,
+	backend types.BackendType,
+) string {
+	columns := parseColumns(entityName, hasUnixMilli, features, backend)
+	return createTableDDL(tableName, columns, pkFields, backend)
 }
 
 func createTableDDL(tableName string, columns ColumnList, pkFields []string, backend types.BackendType) string {
diff --git a/internal/database/offline/bigquery/join.go b/internal/database/offline/bigquery/join.go
index 5bcb999c..e490af22 100644
--- a/internal/database/offline/bigquery/join.go
+++ b/internal/database/offline/bigquery/join.go
@@ -50,7 +50,7 @@ func bigqueryQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string,
 				default:
 				}
 			}
-			defer close(data)
+			close(data)
 		}()
 
 		for {
diff --git a/internal/database/offline/sqlutil/create_table.go b/internal/database/offline/sqlutil/create_table.go
index 3496246f..20358bc2 100644
--- a/internal/database/offline/sqlutil/create_table.go
+++ b/internal/database/offline/sqlutil/create_table.go
@@ -13,33 +13,23 @@ func CreateTable(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.CreateTabl
 	if !supportIndex(dbOpt.Backend) {
 		return createTableNoIndex(ctx, dbOpt, opt)
 	}
-	params := dbutil.BuildTableSchemaParams{
-		TableName:  opt.TableName,
-		EntityName: opt.EntityName,
-		Features:   opt.Features,
-		Backend:    dbOpt.Backend,
-	}
 	switch opt.TableType {
 	case types.TableBatchSnapshot:
 		// Create primary key (entity_key) on batch snapshot table
-		params.PrimaryKeys = []string{opt.EntityName}
-		params.HasUnixMilli = false
-		schema := dbutil.BuildTableSchema(params)
+		pkFields := []string{opt.EntityName}
+		schema := dbutil.BuildTableSchema(opt.TableName, opt.EntityName, false, opt.Features, pkFields, dbOpt.Backend)
 		if err := dbOpt.ExecContext(ctx, schema, nil); err != nil {
 			return err
 		}
 	case types.TableStreamSnapshot:
 		// Create primary key (entity_key) on stream snapshot table
-		params.PrimaryKeys = []string{opt.EntityName}
-		params.HasUnixMilli = true
-		schema := dbutil.BuildTableSchema(params)
+		pkFields := []string{opt.EntityName}
+		schema := dbutil.BuildTableSchema(opt.TableName, opt.EntityName, true, opt.Features, pkFields, dbOpt.Backend)
 		if err := dbOpt.ExecContext(ctx, schema, nil); err != nil {
 			return err
 		}
 	case types.TableStreamCdc:
-		params.PrimaryKeys = nil
-		params.HasUnixMilli = true
-		schema := dbutil.BuildTableSchema(params)
+		schema := dbutil.BuildTableSchema(opt.TableName, opt.EntityName, true, opt.Features, nil, dbOpt.Backend)
 		if err := dbOpt.ExecContext(ctx, schema, nil); err != nil {
 			return err
 		}
@@ -68,16 +58,9 @@ func createTableNoIndex(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.Cre
 
 	tableName := opt.TableName
 	if dbOpt.Backend == types.BackendBigQuery {
-		tableName = fmt.Sprintf("%s.%s", *dbOpt.DatasetID, opt.TableName)
+		tableName = fmt.Sprintf("%s.%s", *dbOpt.DatasetID, tableName)
 	}
-	schema := dbutil.BuildTableSchema(dbutil.BuildTableSchemaParams{
-		TableName:    tableName,
-		EntityName:   opt.EntityName,
-		HasUnixMilli: hasUnixMilli,
-		Features:     opt.Features,
-		PrimaryKeys:  nil,
-		Backend:      dbOpt.Backend,
-	})
+	schema := dbutil.BuildTableSchema(tableName, opt.EntityName, hasUnixMilli, opt.Features, nil, dbOpt.Backend)
 	if err := dbOpt.ExecContext(ctx, schema, nil); err != nil {
 		return err
 	}
diff --git a/internal/database/offline/sqlutil/import.go b/internal/database/offline/sqlutil/import.go
index f4ee2363..b1d98a8b 100644
--- a/internal/database/offline/sqlutil/import.go
+++ b/internal/database/offline/sqlutil/import.go
@@ -16,19 +16,14 @@ type LoadData func(tx *sqlx.Tx, ctx context.Context, opt dbutil.LoadDataFromSour
 
 func Import(ctx context.Context, db *sqlx.DB, opt offline.ImportOpt, loadData LoadData, backend types.BackendType) (int64, error) {
 	var revision int64
-	params := dbutil.BuildTableSchemaParams{
-		TableName:    opt.SnapshotTableName,
-		EntityName:   opt.EntityName,
-		HasUnixMilli: false,
-		Features:     opt.Features,
-		Backend:      backend,
-	}
 	err := dbutil.WithTransaction(db, ctx, func(ctx context.Context, tx *sqlx.Tx) error {
-		// create the data table: streaming snapshot table doesn't have primary key
+		// create the data table
+		var pkFields []string
 		if opt.Category == types.CategoryBatch {
-			params.PrimaryKeys = []string{opt.EntityName}
+			// streaming snapshot table doesn't have primary key
+			pkFields = []string{opt.EntityName}
 		}
-		schema := dbutil.BuildTableSchema(params)
+		schema := dbutil.BuildTableSchema(opt.SnapshotTableName, opt.EntityName, false, opt.Features, pkFields, backend)
 		_, err := tx.ExecContext(ctx, schema)
 		if err != nil {
 			return errdefs.WithStack(err)
diff --git a/internal/database/offline/sqlutil/join.go b/internal/database/offline/sqlutil/join.go
index c40203e7..2ce4ed45 100644
--- a/internal/database/offline/sqlutil/join.go
+++ b/internal/database/offline/sqlutil/join.go
@@ -39,7 +39,7 @@ type DoJoinOpt struct {
 func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.JoinResult, error) {
 	if err := validateJoinOpt(opt); err != nil {
 		data := make(chan types.JoinRecord)
-		defer close(data)
+		close(data)
 		return &types.JoinResult{Data: data}, nil
 	}
 
@@ -54,7 +54,7 @@ func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.Join
 	}
 	if timeRange.MinUnixMilli == nil || timeRange.MaxUnixMilli == nil {
 		data := make(chan types.JoinRecord)
-		defer close(data)
+		close(data)
 		return &types.JoinResult{Data: data}, nil
 	}
 
@@ -212,7 +212,7 @@ type readJoinedTableOpt struct {
 func readJoinedTable(ctx context.Context, dbOpt dbutil.DBOpt, opt readJoinedTableOpt) (*types.JoinResult, error) {
 	if len(opt.TableNames) == 0 {
 		data := make(chan types.JoinRecord)
-		defer close(data)
+		close(data)
 		return &types.JoinResult{Data: data}, nil
 	}
 
@@ -316,8 +316,8 @@ func sqlxQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string, hea
 				}
 			}
 
-			defer rows.Close()
-			defer close(data)
+			rows.Close()
+			close(data)
 		}()
 
 		for rows.Next() {
diff --git a/internal/database/online/sqlutil/create_table.go b/internal/database/online/sqlutil/create_table.go
index 858b0a73..d891495d 100644
--- a/internal/database/online/sqlutil/create_table.go
+++ b/internal/database/online/sqlutil/create_table.go
@@ -15,14 +15,7 @@ func CreateTable(ctx context.Context, dbOpt dbutil.DBOpt, opt online.CreateTable
 		return err
 	}
 	// Step 2: create new table
-	schema := dbutil.BuildTableSchema(dbutil.BuildTableSchemaParams{
-		TableName:    opt.TableName,
-		EntityName:   opt.EntityName,
-		HasUnixMilli: false,
-		Features:     opt.Features,
-		PrimaryKeys:  []string{opt.EntityName},
-		Backend:      dbOpt.Backend,
-	})
+	schema := dbutil.BuildTableSchema(opt.TableName, opt.EntityName, false, opt.Features, []string{opt.EntityName}, dbOpt.Backend)
 	err := dbOpt.ExecContext(ctx, schema, nil)
 	if err != nil {
 		return errdefs.WithStack(err)
diff --git a/internal/database/online/sqlutil/import.go b/internal/database/online/sqlutil/import.go
index c29245e3..e011cd16 100644
--- a/internal/database/online/sqlutil/import.go
+++ b/internal/database/online/sqlutil/import.go
@@ -26,14 +26,7 @@ func Import(ctx context.Context, db *sqlx.DB, opt online.ImportOpt, backend type
 
 	return dbutil.WithTransaction(db, ctx, func(ctx context.Context, tx *sqlx.Tx) error {
 		// create the data table
-		schema := dbutil.BuildTableSchema(dbutil.BuildTableSchemaParams{
-			TableName:    tableName,
-			EntityName:   entity.Name,
-			HasUnixMilli: false,
-			Features:     opt.Features,
-			PrimaryKeys:  []string{entity.Name},
-			Backend:      backend,
-		})
+		schema := dbutil.BuildTableSchema(tableName, entity.Name, false, opt.Features, []string{entity.Name}, backend)
 		_, err := tx.ExecContext(ctx, schema)
 		if err != nil {
 			return errdefs.WithStack(err)
diff --git a/pkg/oomstore/join.go b/pkg/oomstore/join.go
index b5b1f246..bb6adc5d 100644
--- a/pkg/oomstore/join.go
+++ b/pkg/oomstore/join.go
@@ -32,7 +32,7 @@ func (s *OomStore) ChannelJoin(ctx context.Context, opt types.ChannelJoinOpt) (*
 	}
 	if len(features) == 0 {
 		data := make(chan types.JoinRecord)
-		defer close(data)
+		close(data)
 
 		return &types.JoinResult{Data: data}, nil
 	}


Test patch:
diff --git a/internal/database/dbutil/schema_test.go b/internal/database/dbutil/schema_test.go
index 5f487fa2..e11ea001 100644
--- a/internal/database/dbutil/schema_test.go
+++ b/internal/database/dbutil/schema_test.go
@@ -81,14 +81,7 @@ CREATE TABLE "user" (
 
 	for _, c := range cases {
 		t.Run(c.description, func(t *testing.T) {
-			schema := dbutil.BuildTableSchema(dbutil.BuildTableSchemaParams{
-				TableName:    c.tableName,
-				EntityName:   c.entity.Name,
-				HasUnixMilli: false,
-				Features:     c.features,
-				PrimaryKeys:  nil,
-				Backend:      c.backend,
-			})
+			schema := dbutil.BuildTableSchema(c.tableName, c.entity.Name, false, c.features, nil, c.backend)
 			require.Equal(t, strings.TrimSpace(c.want), schema)
 		})
 	}