fix single file target hanging

slingdata-io · Sep 24, 2024 · 2b4c938 · 2b4c938
1 parent be03c9c
commit 2b4c938
Show file tree

Hide file tree

Showing 6 changed files with 22 additions and 8 deletions.
diff --git a/.github/workflows/build-release.yml b/.github/workflows/build-release.yml
@@ -105,7 +105,7 @@ jobs:
         with:
           distribution: goreleaser
           version: 'v1.26.2'
-          args: release --clean --skip=validate -f .goreleaser.mac.yaml
+          args: release --clean --skip=validate -f .goreleaser.mac.yaml --release-notes=.goreleaser.notes.md
 
       - name: Post Tests
         run: |
@@ -155,7 +155,7 @@ jobs:
         with:
           distribution: goreleaser
           version: 'v1.26.2'
-          args: release --clean --skip=validate -f .goreleaser.windows.yaml
+          args: release --clean --skip=validate -f .goreleaser.windows.yaml --release-notes=.goreleaser.notes.md
 
       - name: Post Tests
         shell: pwsh
@@ -198,7 +198,7 @@ jobs:
         with:
           distribution: goreleaser
           version: 'v1.26.2'
-          args: release --clean --skip=validate -f .goreleaser.linux.arm64.yaml
+          args: release --clean --skip=validate -f .goreleaser.linux.arm64.yaml --release-notes=.goreleaser.notes.md
 
       - name: Post Tests
         run: |
@@ -253,7 +253,7 @@ jobs:
         with:
           distribution: goreleaser
           version: 'v1.26.2'
-          args: release --clean --skip=validate -f .goreleaser.linux.amd64.yaml
+          args: release --clean --skip=validate -f .goreleaser.linux.amd64.yaml --release-notes=.goreleaser.notes.md
 
       - name: Post Tests
         run: |

diff --git a/.goreleaser.notes.md b/.goreleaser.notes.md
@@ -0,0 +1 @@
+
diff --git a/cmd/sling/tests/replications/r.16.yaml b/cmd/sling/tests/replications/r.16.yaml
@@ -0,0 +1,9 @@
+source: oracle
+target: local
+
+defaults:
+  mode: full-refresh
+
+streams:
+  system.TEST1K_ORACLE_SQLLDR:
+    object: /tmp/test.csv
diff --git a/cmd/sling/tests/suite.cli.tsv b/cmd/sling/tests/suite.cli.tsv
@@ -47,6 +47,7 @@ n	test_name	rows	bytes	streams	fails	output_contains	command
 46	Run sling with replication configuration 15 incremental	>0		1		incremental	ICEBERG_MODE=incremental sling run -r cmd/sling/tests/replications/r.15.yaml --streams sling_test/lineitem_iceberg/
 47	Run sling with replication configuration 14			5			sling run -r cmd/sling/tests/replications/r.14.yaml
 48	Run sling with replication configuration 14	0					sling run -r cmd/sling/tests/replications/r.14.yaml --streams cmd/sling/tests/files/test1.csv,cmd/sling/tests/files/test1.upsert.csv # file incremental. Second run should have no new rows
-49	Run sling with task configuration	24					sling run -c cmd/sling/tests/task.yaml
-50	Run sling with Parquet source	1018					sling run --src-stream 'file://cmd/sling/tests/files/parquet' --stdout > /dev/null
-51	Run sling with empty input	0				execution succeeded	echo '' | sling run --stdout
+49	Run sling with replication configuration 16	1000		1			sling run -r cmd/sling/tests/replications/r.16.yaml
+50	Run sling with task configuration	24					sling run -c cmd/sling/tests/task.yaml
+51	Run sling with Parquet source	1018					sling run --src-stream 'file://cmd/sling/tests/files/parquet' --stdout > /dev/null
+52	Run sling with empty input	0				execution succeeded	echo '' | sling run --stdout
diff --git a/core/dbio/iop/dataflow.go b/core/dbio/iop/dataflow.go
@@ -741,6 +741,8 @@ func MergeDataflow(df *Dataflow) (dsN *Datastream) {
 				shaper, err := batch.Columns.MakeShaper(dsN.Columns)
 				if err != nil {
 					dsN.Context.CaptureErr(g.Error(err, "could not MakeShaper"))
+					dsN.Close()
+					return
 				}
 				if shaper == nil {
 					shaper = &Shaper{
@@ -766,6 +768,7 @@ func MergeDataflow(df *Dataflow) (dsN *Datastream) {
 	err := dsN.Start()
 	if err != nil {
 		df.Context.CaptureErr(err)
+		dsN.Close()
 	}
 
 	return dsN

diff --git a/core/dbio/iop/datatype.go b/core/dbio/iop/datatype.go
@@ -445,7 +445,7 @@ func (cols Columns) MakeShaper(tgtColumns Columns) (shaper *Shaper, err error) {
 	}
 
 	// determine diff, and match order of target columns
-	tgtColNames := tgtColumns.Names()
+	tgtColNames := tgtColumns.Names(true)
 	diffCols := len(tgtColumns) != len(srcColumns)
 	colMap := map[int]int{}
 	for s, col := range srcColumns {