Feature/issue 48 sequential handle glob match folder (#50)

yokofly · web-flow · commit b470d7474358 · 2025-08-04T20:06:41.000-07:00
diff --git a/core/dbio/filesys/fs.go b/core/dbio/filesys/fs.go
@@ -573,6 +573,8 @@ func (fs *BaseFileSysClient) ReadDataflow(url string, cfg ...iop.FileStreamConfi
 			err = g.Error(err, "Error getting paths")
 			return
 		}
+
+		nodes.Sort()
 	}
 
 	df, err = GetDataflow(fs.Self(), nodes, Cfg)
@@ -1180,26 +1182,8 @@ func MergeReaders(fs FileSysClient, fileType dbio.FileType, nodes FileNodes, lim
 		g.LogError(err)
 	}
 
-	concurrency := runtime.NumCPU()
-	switch {
-	case fs.GetProp("CONCURRENCY") != "":
-		concurrency = cast.ToInt(fs.GetProp("CONCURRENCY"))
-	case g.In(fs.FsType(), dbio.TypeFileS3, dbio.TypeFileGoogle, dbio.TypeFileAzure):
-		switch {
-		// lots of small files (less than 50kb)
-		case len(nodes) > 100 && nodes.AvgSize() < uint64(50*1024):
-			concurrency = 20
-		default:
-			concurrency = 10
-		}
-	case fs.FsType() == dbio.TypeFileLocal:
-		concurrency = 3
-	case concurrency == 1:
-		concurrency = 3
-	}
-
-	g.DebugLow("merging %s readers of %d files [concurrency=%d] from %s", fileType, len(nodes), concurrency, url)
-	readerChn := make(chan *iop.ReaderReady, concurrency)
+	g.DebugLow("merging %s readers of %d files (sequential processing) from %s", fileType, len(nodes), url)
+	readerChn := make(chan *iop.ReaderReady)
 	go func() {
 		defer close(readerChn)
 
@@ -1209,24 +1193,18 @@ func MergeReaders(fs FileSysClient, fileType dbio.FileType, nodes FileNodes, lim
 				continue
 			}
 
-			ds.Context.Wg.Read.Add()
-			go func(path string) {
-				defer ds.Context.Wg.Read.Done()
-				g.Debug("processing reader from %s", path)
+			g.Debug("processing reader from %s", path)
 
-				reader, err := fs.Self().GetReader(path)
-				if err != nil {
-					setError(g.Error(err, "Error getting reader"))
-					return
-				}
+			reader, err := fs.Self().GetReader(path)
+			if err != nil {
+				setError(g.Error(err, "Error getting reader"))
+				return
+			}
 
-				r := &iop.ReaderReady{Reader: reader, URI: path}
-				readerChn <- r
-			}(path)
+			r := &iop.ReaderReady{Reader: reader, URI: path}
+			readerChn <- r
 		}
 
-		ds.Context.Wg.Read.Wait()
-
 	}()
 
 	if g.In(fileType, dbio.FileTypeCsv, dbio.FileTypeJson, dbio.FileTypeJsonLines, dbio.FileTypeXml) {
diff --git a/go.mod b/go.mod
@@ -1,8 +1,8 @@
 module github.com/slingdata-io/sling-cli
 
-go 1.22
+go 1.24
 
-toolchain go1.22.5
+toolchain go1.24.5
 
 require (
 	cloud.google.com/go v0.115.0
