chore: fix remove warehouse jobs panic

warehouse/integrations/azure-synapse/azure_synapse_test.go

@@ -702,7 +702,7 @@ func newMockUploader(
 
  mockUploader := mockuploader.NewMockUploader(ctrl)
  mockUploader.EXPECT().UseRudderStorage().Return(false).AnyTimes()
- mockUploader.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).Return(loadFiles).AnyTimes()
+ mockUploader.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).Return(loadFiles, nil).AnyTimes()
  mockUploader.EXPECT().GetTableSchemaInUpload(tableName).Return(schemaInUpload).AnyTimes()
  mockUploader.EXPECT().GetTableSchemaInWarehouse(tableName).Return(schemaInWarehouse).AnyTimes()
 

warehouse/integrations/bigquery/bigquery.go

@@ -412,11 +412,10 @@ func (bq *BigQuery) loadFileLocations(
  }
  return []warehouseutils.LoadFile{loadfile}, nil
  default:
- metadata := bq.uploader.GetLoadFilesMetadata(
+ return bq.uploader.GetLoadFilesMetadata(
  ctx,
  warehouseutils.GetLoadFilesOptions{Table: tableName},
  )
- return metadata, nil
  }
 }
 

warehouse/integrations/bigquery/bigquery_test.go

@@ -10,19 +10,21 @@ import (
  "testing"
  "time"
 
- "github.com/golang/mock/gomock"
  "golang.org/x/exp/slices"
 
+ "github.com/golang/mock/gomock"
+
  "github.com/rudderlabs/rudder-go-kit/config"
  "github.com/rudderlabs/rudder-go-kit/filemanager"
  "github.com/rudderlabs/rudder-go-kit/logger"
  mockuploader "github.com/rudderlabs/rudder-server/warehouse/internal/mocks/utils"
  "github.com/rudderlabs/rudder-server/warehouse/internal/model"
 
  "cloud.google.com/go/bigquery"
+ "google.golang.org/api/option"
+
  "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/require"
- "google.golang.org/api/option"
 
  "github.com/rudderlabs/compose-test/compose"
  "github.com/rudderlabs/compose-test/testcompose"
@@ -958,8 +960,8 @@ func newMockUploader(
  mockUploader := mockuploader.NewMockUploader(ctrl)
  mockUploader.EXPECT().UseRudderStorage().Return(false).AnyTimes()
  mockUploader.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).DoAndReturn(
- func(ctx context.Context, options warehouseutils.GetLoadFilesOptions) []warehouseutils.LoadFile {
- return slices.Clone(loadFiles)
+ func(ctx context.Context, options warehouseutils.GetLoadFilesOptions) ([]warehouseutils.LoadFile, error) {
+ return slices.Clone(loadFiles), nil
  },
  ).AnyTimes()
  mockUploader.EXPECT().GetTableSchemaInUpload(tableName).Return(schemaInUpload).AnyTimes()

warehouse/integrations/clickhouse/clickhouse_test.go

@@ -1201,7 +1201,7 @@ func newMockUploader(
  u := mockuploader.NewMockUploader(ctrl)
  u.EXPECT().GetSampleLoadFileLocation(gomock.Any(), gomock.Any()).Return(sampleLocation, nil).AnyTimes()
  u.EXPECT().GetTableSchemaInUpload(gomock.Any()).Return(tableSchema).AnyTimes()
- u.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).Return(metadata).AnyTimes()
+ u.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).Return(metadata, nil).AnyTimes()
  u.EXPECT().UseRudderStorage().Return(false).AnyTimes()
  u.EXPECT().IsWarehouseSchemaEmpty().Return(true).AnyTimes()
 

warehouse/integrations/deltalake/deltalake_test.go

@@ -1129,8 +1129,8 @@ func newMockUploader(
  mockUploader.EXPECT().ShouldOnDedupUseNewRecord().Return(onDedupUseNewRecords).AnyTimes()
  mockUploader.EXPECT().CanAppend().Return(canAppend).AnyTimes()
  mockUploader.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).DoAndReturn(
- func(ctx context.Context, options warehouseutils.GetLoadFilesOptions) []warehouseutils.LoadFile {
- return slices.Clone(loadFiles)
+ func(ctx context.Context, options warehouseutils.GetLoadFilesOptions) ([]warehouseutils.LoadFile, error) {
+ return slices.Clone(loadFiles), nil
  },
  ).AnyTimes()
  mockUploader.EXPECT().GetSampleLoadFileLocation(gomock.Any(), gomock.Any()).Return(loadFiles[0].Location, nil).AnyTimes()

warehouse/integrations/mssql/mssql_test.go

@@ -741,7 +741,7 @@ func newMockUploader(
 
  mockUploader := mockuploader.NewMockUploader(ctrl)
  mockUploader.EXPECT().UseRudderStorage().Return(false).AnyTimes()
- mockUploader.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).Return(loadFiles).AnyTimes()
+ mockUploader.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).Return(loadFiles, nil).AnyTimes()
  mockUploader.EXPECT().GetTableSchemaInUpload(tableName).Return(schemaInUpload).AnyTimes()
  mockUploader.EXPECT().GetTableSchemaInWarehouse(tableName).Return(schemaInWarehouse).AnyTimes()
 

warehouse/integrations/postgres/postgres_test.go

@@ -851,7 +851,7 @@ func mockUploader(
 
  mockUploader := mockuploader.NewMockUploader(ctrl)
  mockUploader.EXPECT().UseRudderStorage().Return(false).AnyTimes()
- mockUploader.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).Return(loadFiles).AnyTimes() // Try removing this
+ mockUploader.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).Return(loadFiles, nil).AnyTimes() // Try removing this
  mockUploader.EXPECT().GetTableSchemaInUpload(tableName).Return(schemaInUpload).AnyTimes()
  mockUploader.EXPECT().GetTableSchemaInWarehouse(tableName).Return(schemaInWarehouse).AnyTimes()
 

warehouse/integrations/redshift/redshift.go

@@ -351,12 +351,16 @@
 }
 
 func (rs *Redshift) generateManifest(ctx context.Context, tableName string) (string, error) {
- loadFiles := warehouseutils.GetS3Locations(rs.Uploader.GetLoadFilesMetadata(
+ metadata, err := rs.Uploader.GetLoadFilesMetadata(
  ctx,
  warehouseutils.GetLoadFilesOptions{
  Table: tableName,
  },
- ))
+ )
+ if err != nil {
+ return "", err
+ }
+ loadFiles := warehouseutils.GetS3Locations(metadata)
 
  entries := lo.Map(loadFiles, func(loadFile warehouseutils.LoadFile, index int) s3ManifestEntry {
  manifestEntry := s3ManifestEntry{

warehouse/integrations/redshift/redshift_test.go

@@ -12,9 +12,10 @@ import (
  "testing"
  "time"
 
- "github.com/golang/mock/gomock"
  "golang.org/x/exp/slices"
 
+ "github.com/golang/mock/gomock"
+
  "github.com/rudderlabs/rudder-go-kit/filemanager"
  mockuploader "github.com/rudderlabs/rudder-server/warehouse/internal/mocks/utils"
  "github.com/rudderlabs/rudder-server/warehouse/internal/model"
@@ -983,8 +984,8 @@ func newMockUploader(
  mockUploader := mockuploader.NewMockUploader(ctrl)
  mockUploader.EXPECT().UseRudderStorage().Return(false).AnyTimes()
  mockUploader.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).DoAndReturn(
- func(ctx context.Context, options warehouseutils.GetLoadFilesOptions) []warehouseutils.LoadFile {
- return slices.Clone(loadFiles)
+ func(ctx context.Context, options warehouseutils.GetLoadFilesOptions) ([]warehouseutils.LoadFile, error) {
+ return slices.Clone(loadFiles), nil
  },
  ).AnyTimes()
  mockUploader.EXPECT().GetTableSchemaInUpload(tableName).Return(schemaInUpload).AnyTimes()

warehouse/integrations/snowflake/snowflake_test.go

@@ -12,9 +12,10 @@ import (
  "testing"
  "time"
 
- "github.com/samber/lo"
  "golang.org/x/exp/slices"
 
+ "github.com/samber/lo"
+
  "github.com/rudderlabs/rudder-go-kit/filemanager"
  "github.com/rudderlabs/rudder-server/warehouse/internal/model"
 
@@ -1034,8 +1035,8 @@ func newMockUploader(
  mockUploader.EXPECT().CanAppend().Return(canAppend).AnyTimes()
  mockUploader.EXPECT().ShouldOnDedupUseNewRecord().Return(dedupUseNewRecord).AnyTimes()
  mockUploader.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).DoAndReturn(
- func(ctx context.Context, options whutils.GetLoadFilesOptions) []whutils.LoadFile {
- return slices.Clone(loadFiles)
+ func(ctx context.Context, options whutils.GetLoadFilesOptions) ([]whutils.LoadFile, error) {
+ return slices.Clone(loadFiles), nil
  },
  ).AnyTimes()
  mockUploader.EXPECT().GetSampleLoadFileLocation(gomock.Any(), gomock.Any()).Return(loadFiles[0].Location, nil).AnyTimes()

warehouse/internal/service/loadfiles/downloader/downloader.go

@@ -47,7 +47,10 @@
  fileNamesLock sync.RWMutex
  )
 
- objects := l.uploader.GetLoadFilesMetadata(ctx, warehouseutils.GetLoadFilesOptions{Table: tableName})
+ objects, err := l.uploader.GetLoadFilesMetadata(ctx, warehouseutils.GetLoadFilesOptions{Table: tableName})
+ if err != nil {
+ return nil, fmt.Errorf("getting load files metadata: %w", err)
+ }
  storageProvider := warehouseutils.ObjectStorageType(
  l.warehouse.Destination.DestinationDefinition.Name,
  l.warehouse.Destination.Config,

warehouse/internal/service/loadfiles/downloader/downloader_test.go

@@ -175,7 +175,7 @@ func TestDownloader(t *testing.T) {
 func newMockUploader(t testing.TB, loadFiles []warehouseutils.LoadFile) *mockuploader.MockUploader {
  ctrl := gomock.NewController(t)
  u := mockuploader.NewMockUploader(ctrl)
- u.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).AnyTimes().Return(loadFiles)
+ u.EXPECT().GetLoadFilesMetadata(gomock.Any(), gomock.Any()).AnyTimes().Return(loadFiles, nil)
  u.EXPECT().UseRudderStorage().Return(false).AnyTimes()
  return u
 }

warehouse/jobs/jobs.go

@@ -29,8 +29,8 @@
  return model.TableSchema{}
 }
 
-func (*WhAsyncJob) GetLoadFilesMetadata(context.Context, warehouseutils.GetLoadFilesOptions) []warehouseutils.LoadFile {
- return []warehouseutils.LoadFile{}
+func (*WhAsyncJob) GetLoadFilesMetadata(context.Context, warehouseutils.GetLoadFilesOptions) ([]warehouseutils.LoadFile, error) {
+ return []warehouseutils.LoadFile{}, nil
 }
 
 func (*WhAsyncJob) GetSampleLoadFileLocation(context.Context, string) (string, error) {

warehouse/router/upload.go

@@ -12,9 +12,10 @@
  "sync/atomic"
  "time"
 
+ "golang.org/x/exp/slices"
+
  "github.com/cenkalti/backoff/v4"
  "github.com/samber/lo"
- "golang.org/x/exp/slices"
 
  "github.com/rudderlabs/rudder-go-kit/config"
  "github.com/rudderlabs/rudder-go-kit/logger"
@@ -283,7 +284,7 @@
 
  marshalledSchema, err := json.Marshal(uploadSchema)
  if err != nil {
- panic(err)
+ return err
  }
 
  err = job.setUploadColumns(UploadColumnsOpts{Fields: []UploadColumn{
@@ -1321,18 +1322,18 @@
 
 // getNewTimings appends current status with current time to timings column
 // e.g. status: exported_data, timings: [{exporting_data: 2020-04-21 15:16:19.687716}] -> [{exporting_data: 2020-04-21 15:16:19.687716, exported_data: 2020-04-21 15:26:34.344356}]
-func (job *UploadJob) getNewTimings(status string) ([]byte, model.Timings) {
+func (job *UploadJob) getNewTimings(status string) ([]byte, model.Timings, error) {
  timings, err := repo.NewUploads(job.db).UploadTimings(job.ctx, job.upload.ID)
  if err != nil {
- job.logger.Error("error getting timing, scrapping them", err)
+ return nil, nil, err
  }
  timing := map[string]time.Time{status: job.now()}
  timings = append(timings, timing)
  marshalledTimings, err := json.Marshal(timings)
  if err != nil {
- panic(err)
+ return nil, nil, err
  }
- return marshalledTimings, timings
+ return marshalledTimings, timings, nil
 }
 
 func (job *UploadJob) getUploadFirstAttemptTime() (timing time.Time) {
@@ -1371,7 +1372,10 @@
  }()
 
  // TODO: fetch upload model instead of just timings
- marshalledTimings, timings := job.getNewTimings(statusOpts.Status)
+ marshalledTimings, timings, err := job.getNewTimings(statusOpts.Status)
+ if err != nil {
+ return
+ }
  opts := []UploadColumn{
  {Column: UploadStatusField, Value: statusOpts.Status},
  {Column: UploadTimingsField, Value: marshalledTimings},
@@ -1792,7 +1796,7 @@
  return true, nil
 }
 
-func (job *UploadJob) GetLoadFilesMetadata(ctx context.Context, options whutils.GetLoadFilesOptions) (loadFiles []whutils.LoadFile) {
+func (job *UploadJob) GetLoadFilesMetadata(ctx context.Context, options whutils.GetLoadFilesOptions) (loadFiles []whutils.LoadFile, err error) {
  var tableFilterSQL string
  if options.Table != "" {
  tableFilterSQL = fmt.Sprintf(` AND table_name='%s'`, options.Table)
@@ -1837,7 +1841,7 @@
  job.logger.Debugf(`Fetching loadFileLocations: %v`, sqlStatement)
  rows, err := job.db.QueryContext(ctx, sqlStatement)
  if err != nil {
- panic(fmt.Errorf("query: %s\nfailed with Error : %w", sqlStatement, err))
+ return nil, fmt.Errorf("query: %s\nfailed with Error : %w", sqlStatement, err)
  }
  defer func() { _ = rows.Close() }()
 
@@ -1846,21 +1850,24 @@
  var metadata json.RawMessage
  err := rows.Scan(&location, &metadata)
  if err != nil {
- panic(fmt.Errorf("failed to scan result from query: %s\nwith Error : %w", sqlStatement, err))
+ return nil, fmt.Errorf("failed to scan result from query: %s\nwith Error : %w", sqlStatement, err)
  }
  loadFiles = append(loadFiles, whutils.LoadFile{
  Location: location,
  Metadata: metadata,
  })
  }
  if err = rows.Err(); err != nil {
- panic(fmt.Errorf("iterate query results: %s\nwith Error : %w", sqlStatement, err))
+ return nil, fmt.Errorf("iterate query results: %s\nwith Error : %w", sqlStatement, err)
  }
  return
 }
 
 func (job *UploadJob) GetSampleLoadFileLocation(ctx context.Context, tableName string) (location string, err error) {
- locations := job.GetLoadFilesMetadata(ctx, whutils.GetLoadFilesOptions{Table: tableName, Limit: 1})
+ locations, err := job.GetLoadFilesMetadata(ctx, whutils.GetLoadFilesOptions{Table: tableName, Limit: 1})
+ if err != nil {
+ return "", fmt.Errorf("get load file metadata: %w", err)
+ }
  if len(locations) == 0 {
  return "", fmt.Errorf(`no load file found for table:%s`, tableName)
  }
@@ -2037,11 +2044,14 @@
 
  // Refresh partitions if exists
  for tableName := range job.upload.UploadSchema {
- loadFiles := job.GetLoadFilesMetadata(job.ctx, whutils.GetLoadFilesOptions{
+ loadFiles, err := job.GetLoadFilesMetadata(job.ctx, whutils.GetLoadFilesOptions{
  Table: tableName,
  StartID: loadFileStartID,
  EndID: loadFileEndID,
  })
+ if err != nil {
+ return fmt.Errorf("get load files metadata: %w", err)
+ }
  batches := lo.Chunk(loadFiles, job.config.refreshPartitionBatchSize)
  for _, batch := range batches {
  if err = repository.RefreshPartitions(job.ctx, tableName, batch); err != nil {

warehouse/utils/utils.go

@@ -237,7 +237,7 @@ type Uploader interface {
  UpdateLocalSchema(ctx context.Context, schema model.Schema) error
  GetTableSchemaInWarehouse(tableName string) model.TableSchema
  GetTableSchemaInUpload(tableName string) model.TableSchema
- GetLoadFilesMetadata(ctx context.Context, options GetLoadFilesOptions) []LoadFile
+ GetLoadFilesMetadata(ctx context.Context, options GetLoadFilesOptions) ([]LoadFile, error)
  GetSampleLoadFileLocation(ctx context.Context, tableName string) (string, error)
  GetSingleLoadFile(ctx context.Context, tableName string) (LoadFile, error)
  ShouldOnDedupUseNewRecord() bool

warehouse/validations/validate.go

@@ -565,8 +565,8 @@
  return "", nil
 }
 
-func (*dummyUploader) GetLoadFilesMetadata(context.Context, warehouseutils.GetLoadFilesOptions) []warehouseutils.LoadFile {
- return nil
+func (*dummyUploader) GetLoadFilesMetadata(context.Context, warehouseutils.GetLoadFilesOptions) ([]warehouseutils.LoadFile, error) {
+ return nil, nil
 }
 
 func (*dummyUploader) GetSingleLoadFile(context.Context, string) (warehouseutils.LoadFile, error) {