Skip to content

feat: add comment #5772

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 5 commits into
base: war-461-create-uploadv2-jobs-final
Choose a base branch
from
Open

feat: add comment #5772

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
fe96a0d
feat: add comment
shekhar-rudder Apr 24, 2025
29da5a8
logs
shekhar-rudder Apr 25, 2025
bf44a28
more logs
shekhar-rudder Apr 25, 2025
41c5bcb
remove load files deletion
shekhar-rudder Apr 25, 2025
7957f80
fix: revert
shekhar-rudder Apr 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions warehouse/internal/loadfiles/loadfiles.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -211,6 +211,7 @@ func (lf *LoadFileGenerator) createFromStaging(ctx context.Context, job *model.U
}()

if !lf.AllowUploadV2JobCreation(job) {
lf.Logger.Infof("V2 job creation disabled. Processing %d staging files", len(toProcessStagingFiles))
err = lf.createUploadJobs(ctx, job, toProcessStagingFiles, publishBatchSize, uniqueLoadGenID)
if err != nil {
return 0, 0, fmt.Errorf("creating upload jobs: %w", err)
Expand All @@ -234,11 +235,13 @@ func (lf *LoadFileGenerator) createFromStaging(ctx context.Context, job *model.U
g, gCtx := errgroup.WithContext(ctx)
if len(v1Files) > 0 {
g.Go(func() error {
lf.Logger.Infof("V1 job creation enabled. Processing %d v1 staging files", len(v1Files))
return lf.createUploadJobs(gCtx, job, v1Files, publishBatchSize, uniqueLoadGenID)
})
}
if len(v2Files) > 0 {
g.Go(func() error {
lf.Logger.Infof("V1 job creation enabled. Processing %d v2 staging files", len(v2Files))
return lf.createUploadV2Jobs(gCtx, job, v2Files, publishBatchSize, uniqueLoadGenID)
})
}
Expand Down Expand Up @@ -329,7 +332,7 @@ func (lf *LoadFileGenerator) publishToNotifier(

destID := job.Upload.DestinationID
destType := job.Upload.DestinationType
lf.Logger.Infof("[WH]: Publishing %d staging files for %s:%s to notifier", len(messages), destType, destID)
lf.Logger.Infof("[WH]: Publishing %d jobs for %s:%s to notifier", len(messages), destType, destID)

ch, err := lf.Notifier.Publish(ctx, &notifier.PublishRequest{
Payloads: messages,
Expand Down Expand Up @@ -401,7 +404,6 @@ func (lf *LoadFileGenerator) processNotifierResponse(ctx context.Context, ch <-c
return nil
}

// Unlike upload type job, for v2 we are not setting the status of staging files
func (lf *LoadFileGenerator) processNotifierResponseV2(ctx context.Context, ch <-chan *notifier.PublishResponse, job *model.UploadJob, chunk []*model.StagingFile) error {
responses, ok := <-ch
if !ok {
Expand Down Expand Up @@ -485,8 +487,9 @@ func (lf *LoadFileGenerator) createUploadV2Jobs(ctx context.Context, job *model.
}
g, gCtx := errgroup.WithContext(ctx)
stagingFileGroups := lf.GroupStagingFiles(stagingFiles, lf.Conf.GetInt("Warehouse.loadFiles.maxSizeInMB", 128))
for _, fileGroups := range lo.Chunk(stagingFileGroups, publishBatchSize) {
for _, group := range fileGroups {
for i, fileGroups := range lo.Chunk(stagingFileGroups, publishBatchSize) {
for j, group := range fileGroups {
lf.Logger.Infof("chunk %d, group %d, size %d", i, j, len(group))
baseReq := lf.generateBaseRequest(job, uniqueLoadGenID, group[0], destinationRevisionIDMap)

stagingFileInfos := make([]StagingFileInfo, len(group))
Expand Down Expand Up @@ -667,6 +670,7 @@ func (lf *LoadFileGenerator) groupBySize(files []*fileWithMaxSize, maxSizeMB int
newSize := batchTableSizes[tableName] + size
if newSize > maxSizeBytes {
canAdd = false
lf.Logger.Infof("Can't add file %d, current size: %d, maxSize: %d", files[i].file.ID, newSize, maxSizeBytes)
break
}
}
Expand All @@ -677,6 +681,7 @@ func (lf *LoadFileGenerator) groupBySize(files []*fileWithMaxSize, maxSizeMB int
for tableName, size := range files[i].file.BytesPerTable {
batchTableSizes[tableName] += size
}
lf.Logger.Infof("Added file %d to batch. New size: %v", files[i].file.ID, batchTableSizes)
// Remove the file from remaining by moving the last element
files[i] = files[len(files)-1]
files = files[:len(files)-1]
Expand All @@ -688,6 +693,7 @@ func (lf *LoadFileGenerator) groupBySize(files []*fileWithMaxSize, maxSizeMB int
// If we couldn't add any files to the batch, add the first file anyway
// This ensures we make progress even with files larger than maxSizeBytes
if len(currentBatch) == 0 {
lf.Logger.Infof("Adding only file %d to batch", files[0].file.ID)
currentBatch = append(currentBatch, files[0].file)
files = files[1:]
}
Expand Down
3 changes: 2 additions & 1 deletion warehouse/router/upload.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ type UploadJobFactory struct {
encodingFactory *encoding.Factory
}

// add comment
type UploadJob struct {
ctx context.Context
db *sqlquerywrapper.DB
Expand Down Expand Up @@ -416,7 +417,7 @@ func (job *UploadJob) run() (err error) {
job.timerStat(nextUploadState.inProgress).SendTiming(time.Since(stateStartTime))

if newStatus == model.ExportedData {
_ = job.loadFilesRepo.Delete(job.ctx, job.upload.ID, job.stagingFileIDs)
// _ = job.loadFilesRepo.Delete(job.ctx, job.upload.ID, job.stagingFileIDs)
break
}

Expand Down
Loading