fix: Async file close and S3 upload

writers/parquet/parquet.go

@@ -229,15 +229,15 @@
 	}
 
 	for basePath, parquetFiles := range p.partitionedFiles {
-		for _, fileMetadata := range parquetFiles {
-			// TODO: Async file close and S3 upload (Good First Issue)
+		ctx := context.Background()
+		err := utils.Concurrent(ctx, parquetFiles, len(parquetFiles), func(_ context.Context, fileMetadata FileMetadata, _ int) error {
 			// construct full file path
 			filePath := filepath.Join(p.config.Path, basePath, fileMetadata.fileName)
 
 			// Remove empty files
 			if fileMetadata.recordCount == 0 {
 				removeLocalFile(filePath, "no records written", fileMetadata.recordCount)
-				continue
+				return nil
 			}
 
 			// Close writers
@@ -286,6 +286,10 @@
 				removeLocalFile(filePath, "uploaded to S3", fileMetadata.recordCount)
 				logger.Infof("Successfully uploaded file to S3: s3://%s/%s", p.config.Bucket, s3KeyPath)
 			}
+			return nil
+		})
+		if err != nil {
+			return fmt.Errorf("failed to close writers and files: %s", err)
 		}
 	}
 	return nil
@@ -350,7 +354,7 @@
 						switch granularity {
 						case "HH":
 							value = timestamp.UTC().Hour()
 						case "DD":
 							value = timestamp.UTC().Day()
 						case "WW":
 							_, value = timestamp.UTC().ISOWeek()
@@ -366,6 +370,7 @@
 			}
 			return fmt.Sprintf("%v", value)
 		}
+
 		if colName == "now()" {
 			return granularityFunction(time.Now().UTC())
 		}