File tree Expand file tree Collapse file tree 2 files changed +37
-10
lines changed
src/core/etl/src/Flow/ETL Expand file tree Collapse file tree 2 files changed +37
-10
lines changed Original file line number Diff line number Diff line change @@ -19,34 +19,53 @@ public function __construct(
19
19
20
20
public function extract (FlowContext $ context ) : \Generator
21
21
{
22
- $ chunk = new Rows ();
22
+ $ chunk = [];
23
+ $ chunkSize = 0 ;
23
24
24
25
foreach ($ this ->extractor ->extract ($ context ) as $ rows ) {
25
26
foreach ($ rows ->chunks ($ this ->chunkSize ) as $ rowsChunk ) {
26
- $ chunk = $ chunk ->merge ($ rowsChunk );
27
+ $ chunk [] = $ rowsChunk ->all ();
28
+ $ chunkSize += $ rowsChunk ->count ();
27
29
28
- if ($ chunk ->count () === $ this ->chunkSize ) {
29
- $ signal = yield $ chunk ;
30
+ if ($ chunkSize === $ this ->chunkSize ) {
31
+ $ signal = yield new Rows (
32
+ ...\array_merge (
33
+ ...$ chunk
34
+ )
35
+ );
30
36
31
37
if ($ signal === Signal::STOP ) {
32
38
return ;
33
39
}
34
- $ chunk = new Rows ();
40
+ $ chunkSize = 0 ;
41
+ $ chunk = [];
35
42
}
36
43
37
- if ($ chunk ->count () > $ this ->chunkSize ) {
38
- $ signal = yield $ chunk ->dropRight ($ chunk ->count () - $ this ->chunkSize );
44
+ if ($ chunkSize > $ this ->chunkSize ) {
45
+ $ allRows = new Rows (
46
+ ...\array_merge (
47
+ ...$ chunk
48
+ )
49
+ );
50
+
51
+ $ signal = yield $ allRows ->dropRight ($ allRows ->count () - $ this ->chunkSize );
39
52
40
53
if ($ signal === Signal::STOP ) {
41
54
return ;
42
55
}
43
- $ chunk = $ chunk ->takeRight ($ chunk ->count () - $ this ->chunkSize );
56
+ $ leftover = $ allRows ->takeRight ($ allRows ->count () - $ this ->chunkSize );
57
+ $ chunk = [$ leftover ->all ()];
58
+ $ chunkSize = $ leftover ->count ();
44
59
}
45
60
}
46
61
}
47
62
48
- if ($ chunk ->count ()) {
49
- yield $ chunk ;
63
+ if ($ chunkSize ) {
64
+ yield new Rows (
65
+ ...\array_merge (
66
+ ...$ chunk
67
+ )
68
+ );
50
69
}
51
70
}
52
71
Original file line number Diff line number Diff line change @@ -61,6 +61,14 @@ public function add(Row ...$rows) : self
61
61
);
62
62
}
63
63
64
+ /**
65
+ * return array<Row>.
66
+ */
67
+ public function all () : array
68
+ {
69
+ return $ this ->rows ;
70
+ }
71
+
64
72
/**
65
73
* @param int<1, max> $size
66
74
*
You can’t perform that action at this time.
0 commit comments