bulker: added temporaryBatchSize option to enable filling tmp table i…

…n batches

sync-sidecar: use temporaryBatchSize=100000 for full-refresh streams.
bulker: redshift: use json_parse_truncate_strings=ON to load table

.github/workflows/test-build-image.yml

@@ -30,7 +30,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v3
       with:
-        go-version: 1.22.5
+        go-version: 1.23.1
         cache: false
 
     - name: Bulker Test

admin/go.mod

@@ -1,8 +1,8 @@
 module github.com/jitsucom/bulker/admin
 
-go 1.22.0
+go 1.23
 
-toolchain go1.22.4
+toolchain go1.23.1
 
 require (
 	github.com/confluentinc/confluent-kafka-go/v2 v2.5.0

all.Dockerfile

@@ -7,7 +7,7 @@ ENV TZ=UTC
 
 WORKDIR /app
 
-FROM golang:1.22.5-bookworm as builder
+FROM golang:1.23.1-bookworm as builder
 
 ARG VERSION
 ENV VERSION $VERSION

bulker.Dockerfile

@@ -5,7 +5,7 @@ RUN apt-get install -y ca-certificates curl
 
 ENV TZ=UTC
 
-FROM golang:1.22.2-bullseye as build
+FROM golang:1.23.1-bullseye as build
 
 ARG VERSION
 ENV VERSION $VERSION

bulkerapp/go.mod

@@ -1,8 +1,8 @@
 module github.com/jitsucom/bulker/bulkerapp
 
-go 1.22.0
+go 1.23
 
-toolchain go1.22.4
+toolchain go1.23.1
 
 require (
 	github.com/confluentinc/confluent-kafka-go/v2 v2.5.0

bulkerlib/go.mod

@@ -1,8 +1,8 @@
 module github.com/jitsucom/bulker/bulkerlib
 
-go 1.22.0
+go 1.23
 
-toolchain go1.22.4
+toolchain go1.23.1
 
 require (
 	cloud.google.com/go v0.115.0

bulkerlib/implementations/file_storage/abstract.go

@@ -228,7 +228,7 @@ func (ps *AbstractFileStorageStream) flushBatchFile(ctx context.Context) (err er
 				_ = file.Close()
 			}()
 			scanner := bufio.NewScanner(file)
-			scanner.Buffer(make([]byte, 1024*10), 1024*1024)
+			scanner.Buffer(make([]byte, 1024*10), 1024*1024*10)
 			i := 0
 			for scanner.Scan() {
 				if !ps.batchFileSkipLines.Contains(i) {

bulkerlib/implementations/sql/abstract_transactional.go

@@ -28,6 +28,8 @@ type AbstractTransactionalSQLStream struct {
 	//function that generate tmp table schema based on target table schema
 	tmpTableFunc       func(ctx context.Context, tableForObject *Table, object types.Object) (table *Table)
 	dstTable           *Table
+	temporaryBatchSize int
+	localBatchFileName string
 	batchFile          *os.File
 	marshaller         types.Marshaller
 	targetMarshaller   types.Marshaller
@@ -62,9 +64,28 @@ func newAbstractTransactionalStream(id string, p SQLAdapter, tableName string, m
 			ps.useDiscriminator = true
 		}
 	}
+	ps.localBatchFileName = localBatchFileOption.Get(&ps.options)
+	ps.temporaryBatchSize = TemporaryBatchSizeOption.Get(&ps.options)
 	return &ps, nil
 }
 
+func (ps *AbstractTransactionalSQLStream) initTmpFile(ctx context.Context) (err error) {
+	if ps.batchFile == nil {
+		if !ps.merge && ps.sqlAdapter.GetBatchFileFormat() == types.FileFormatNDJSON {
+			//without merge we can write file with compression - no need to convert
+			ps.marshaller, _ = types.NewMarshaller(ps.sqlAdapter.GetBatchFileFormat(), ps.sqlAdapter.GetBatchFileCompression())
+		} else {
+			ps.marshaller, _ = types.NewMarshaller(types.FileFormatNDJSON, types.FileCompressionNONE)
+		}
+		ps.targetMarshaller, err = types.NewMarshaller(ps.sqlAdapter.GetBatchFileFormat(), ps.sqlAdapter.GetBatchFileCompression())
+		if err != nil {
+			return err
+		}
+		ps.batchFile, err = os.CreateTemp("", ps.localBatchFileName+"_*"+ps.marshaller.FileExtension())
+	}
+	return
+}
+
 func (ps *AbstractTransactionalSQLStream) init(ctx context.Context) (err error) {
 	if ps.inited {
 		return nil
@@ -77,20 +98,10 @@ func (ps *AbstractTransactionalSQLStream) init(ctx context.Context) (err error)
 			return fmt.Errorf("failed to setup s3 client: %v", err)
 		}
 	}
-	localBatchFile := localBatchFileOption.Get(&ps.options)
-	if localBatchFile != "" && ps.batchFile == nil {
-		ps.marshaller, _ = types.NewMarshaller(types.FileFormatNDJSON, types.FileCompressionNONE)
-		ps.targetMarshaller, err = types.NewMarshaller(ps.sqlAdapter.GetBatchFileFormat(), ps.sqlAdapter.GetBatchFileCompression())
+	if ps.localBatchFileName != "" && ps.batchFile == nil {
+		err = ps.initTmpFile(ctx)
 		if err != nil {
-			return err
-		}
-		if !ps.merge && ps.sqlAdapter.GetBatchFileFormat() == types.FileFormatNDJSON {
-			//without merge we can write file with compression - no need to convert
-			ps.marshaller, _ = types.NewMarshaller(ps.sqlAdapter.GetBatchFileFormat(), ps.sqlAdapter.GetBatchFileCompression())
-		}
-		ps.batchFile, err = os.CreateTemp("", localBatchFile+"_*"+ps.marshaller.FileExtension())
-		if err != nil {
-			return err
+			return
 		}
 	}
 	err = ps.AbstractSQLStream.init(ctx)
@@ -146,9 +157,11 @@ func (ps *AbstractTransactionalSQLStream) flushBatchFile(ctx context.Context) (s
 		}
 		_ = ps.batchFile.Close()
 		_ = os.Remove(ps.batchFile.Name())
+		ps.batchFile = nil
+		ps.eventsInBatch = 0
 	}()
-	if ps.eventsInBatch > 0 {
-		err = ps.tx.CreateTable(ctx, table)
+	if ps.batchFile != nil && ps.eventsInBatch > 0 {
+		_, err = ps.sqlAdapter.TableHelper().EnsureTableWithCaching(ctx, ps.tx, ps.id, table)
 		if err != nil {
 			return state, errorj.Decorate(err, "failed to create table")
 		}
@@ -345,6 +358,17 @@ func (ps *AbstractTransactionalSQLStream) flushBatchFile(ctx context.Context) (s
 //}
 
 func (ps *AbstractTransactionalSQLStream) writeToBatchFile(ctx context.Context, targetTable *Table, processedObject types.Object) error {
+	if ps.temporaryBatchSize > 0 && ps.eventsInBatch >= ps.temporaryBatchSize {
+		ws, err := ps.flushBatchFile(ctx)
+		ps.state.AddWarehouseState(ws)
+		if err != nil {
+			return err
+		}
+		err = ps.initTmpFile(ctx)
+		if err != nil {
+			return err
+		}
+	}
 	ps.adjustTables(ctx, targetTable, processedObject)
 	ps.updateRepresentationTable(ps.tmpTable)
 	err := ps.marshaller.InitSchema(ps.batchFile, nil, nil)
@@ -435,7 +459,7 @@ func (ps *AbstractTransactionalSQLStream) Consume(ctx context.Context, object ty
 	if err != nil {
 		return
 	}
-	batchFile := ps.batchFile != nil
+	batchFile := ps.localBatchFileName != ""
 	if batchFile {
 		err = ps.writeToBatchFile(ctx, tableForObject, processedObject)
 	} else {

bulkerlib/implementations/sql/bigdata_test.go

@@ -31,11 +31,9 @@ func TestMillionRows(t *testing.T) {
 	}
 	tests := []bulkerTestConfig{
 		{
-			name:  "one_million_rows",
-			modes: []bulker.BulkMode{bulker.Batch},
-			expectedTable: ExpectedTable{
-				Columns: justColumns("_timestamp", "id", "name"),
-			},
+			name:              "one_million_rows",
+			modes:             []bulker.BulkMode{bulker.Batch},
+			batchSize:         eventsCount + 1,
 			expectedRowsCount: eventsCount,
 			configIds:         configIds,
 		},
@@ -80,6 +78,38 @@ func TestMillionRowsBatched(t *testing.T) {
 		})
 	}
 }
+
+func TestMillionRowsTmpBatches(t *testing.T) {
+	configsEnabled := os.Getenv("BULKER_TEST_MILLION_ROWS_TMP_BATCHES")
+	if configsEnabled == "" {
+		t.Skip("This test is disabled by default. To enable it set BULKER_TEST_MILLION_ROWS env variable with comma separated list of bulker config ids")
+		return
+	}
+	configIds := strings.Split(configsEnabled, ",")
+	configIds = utils.ArrayIntersection(allBulkerConfigs, configIds)
+	if len(configIds) == 0 {
+		t.Skipf("Test was skipped. IDs: %v is not among configured configs: %v", configIds, allBulkerConfigs)
+		return
+	}
+	tests := []bulkerTestConfig{
+		{
+			name:              "one_million_rows_tmp_batches",
+			modes:             []bulker.BulkMode{bulker.Batch},
+			batchSize:         eventsCount + 1,
+			expectedRowsCount: eventsCount,
+			configIds:         configIds,
+			streamOptions:     []bulker.StreamOption{WithTemporaryBatchSize(34000)},
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			runTestConfig(t, tt, testLotOfEvents)
+		})
+	}
+}
+
 func testLotOfEvents(t *testing.T, testConfig bulkerTestConfig, mode bulker.BulkMode) {
 	reqr := require.New(t)
 	blk, err := bulker.CreateBulker(*testConfig.config)
@@ -149,7 +179,7 @@ func testLotOfEvents(t *testing.T, testConfig bulkerTestConfig, mode bulker.Bulk
 	state, err := stream.Complete(ctx)
 	sqlAdapter = blk.(SQLAdapter)
 	PostStep("stream_complete", testConfig, mode, reqr, err)
-	logging.Infof("%d. batch is completed in %s", i, time.Since(startTime))
+	logging.Infof("%d. batch is completed in %s: state: %s", i, time.Since(startTime), state.String())
 
 	if testConfig.expectedState != nil {
 		reqr.Equal(*testConfig.expectedState, state)

bulkerlib/implementations/sql/bulker_test.go

@@ -531,6 +531,7 @@ func testStream(t *testing.T, testConfig bulkerTestConfig, mode bulker.BulkMode)
 	scanner := bufio.NewScanner(file)
 	i := 0
 	streamNum := 0
+	scanner.Buffer(make([]byte, 1024*10), 1024*1024*10)
 	for scanner.Scan() {
 		if i > 0 && testConfig.batchSize > 0 && i%testConfig.batchSize == 0 {
 			_, err := stream.Complete(ctx)

bulkerlib/implementations/sql/options.go

@@ -60,6 +60,12 @@ var (
 		ParseFunc:    utils.ParseInt,
 	}
 
+	TemporaryBatchSizeOption = bulker.ImplementationOption[int]{
+		Key:          "temporaryBatchSize",
+		DefaultValue: 0,
+		ParseFunc:    utils.ParseInt,
+	}
+
 	localBatchFileOption = bulker.ImplementationOption[string]{Key: "BULKER_OPTION_LOCAL_BATCH_FILE"}
 
 	s3BatchFileOption = bulker.ImplementationOption[*S3OptionConfig]{Key: "BULKER_OPTION_S3_BATCH_FILE"}
@@ -71,7 +77,7 @@ func init() {
 	bulker.RegisterOption(&OmitNilsOption)
 	bulker.RegisterOption(&SchemaFreezeOption)
 	bulker.RegisterOption(&MaxColumnsCount)
-
+	bulker.RegisterOption(&TemporaryBatchSizeOption)
 }
 
 type S3OptionConfig struct {
@@ -142,3 +148,9 @@ func withS3BatchFile(s3OptionConfig *S3OptionConfig) bulker.StreamOption {
 		s3BatchFileOption.Set(options, s3OptionConfig)
 	}
 }
+
+func WithTemporaryBatchSize(temporaryBatchSize int) bulker.StreamOption {
+	return func(options *bulker.StreamOptions) {
+		TemporaryBatchSizeOption.Set(options, temporaryBatchSize)
+	}
+}

bulkerlib/implementations/sql/redshift.go

@@ -197,6 +197,7 @@ func (p *Redshift) LoadTable(ctx context.Context, targetTable *Table, loadSource
 	if s3Config.Folder != "" {
 		fileKey = s3Config.Folder + "/" + fileKey
 	}
+	_, _ = p.txOrDb(ctx).ExecContext(ctx, "SET json_parse_truncate_strings=ON")
 	statement := fmt.Sprintf(redshiftCopyTemplate, namespace, quotedTableName, strings.Join(columnNames, ","), s3Config.Bucket, fileKey, s3Config.AccessKeyID, s3Config.SecretKey, s3Config.Region)
 	if _, err := p.txOrDb(ctx).ExecContext(ctx, statement); err != nil {
 		return state, errorj.CopyError.Wrap(err, "failed to copy data from s3").

bulkerlib/implementations/sql/replacepartition_stream.go

@@ -87,7 +87,7 @@ func (ps *ReplacePartitionStream) Complete(ctx context.Context) (state bulker.St
 		}
 		err = ps.clearPartition(ctx, ps.tx)
 		if err == nil && ps.state.SuccessfulRows > 0 {
-			if ps.batchFile != nil {
+			if ps.localBatchFileName != "" {
 				ws, err := ps.flushBatchFile(ctx)
 				ps.state.AddWarehouseState(ws)
 				if err != nil {

bulkerlib/implementations/sql/replacetable_stream.go

@@ -54,7 +54,7 @@ func (ps *ReplaceTableStream) Complete(ctx context.Context) (state bulker.State,
 	if ps.state.LastError == nil {
 		//if at least one object was inserted
 		if ps.state.SuccessfulRows > 0 {
-			if ps.batchFile != nil {
+			if ps.localBatchFileName != "" {
 				ws, err := ps.flushBatchFile(ctx)
 				ps.state.AddWarehouseState(ws)
 				if err != nil {

bulkerlib/implementations/sql/transactional_stream.go

@@ -62,7 +62,7 @@ func (ps *TransactionalStream) Complete(ctx context.Context) (state bulker.State
 			Name:            "consume",
 			TimeProcessedMs: time.Since(ps.startTime).Milliseconds(),
 		})
-		if ps.batchFile != nil {
+		if ps.localBatchFileName != "" {
 			ws, err := ps.flushBatchFile(ctx)
 			ps.state.AddWarehouseState(ws)
 			if err != nil {

connectors/airbytecdk/go.mod

@@ -1,3 +1,3 @@
 module github.com/jitsucom/bulker/airbytecdk
 
-go 1.22
+go 1.23

connectors/firebase.Dockerfile

@@ -5,7 +5,7 @@ RUN apt-get install -y ca-certificates curl
 
 ENV TZ=UTC
 
-FROM golang:1.22.2-bullseye as build
+FROM golang:1.23.1-bullseye as build
 
 RUN apt-get install gcc libc6-dev
 

connectors/firebase/go.mod

@@ -1,6 +1,6 @@
 module github.com/jitsucom/bulker/connectors/firebase
 
-go 1.22
+go 1.23
 
 require (
 	cloud.google.com/go/firestore v1.15.0

eventslog/go.mod

@@ -1,6 +1,6 @@
 module github.com/jitsucom/bulker/eventslog
 
-go 1.22
+go 1.23
 
 require (
 	github.com/ClickHouse/clickhouse-go/v2 v2.26.0

go.work

@@ -1,6 +1,6 @@
-go 1.22.0
+go 1.23
 
-toolchain go1.22.4
+toolchain go1.23.1
 
 use (
 	./bulkerapp

ingest.Dockerfile

@@ -5,7 +5,7 @@ RUN apt-get install -y ca-certificates curl
 
 ENV TZ=UTC
 
-FROM golang:1.22.2-bullseye as build
+FROM golang:1.23.1-bullseye as build
 
 ARG VERSION
 ENV VERSION $VERSION

ingest/go.mod

@@ -1,8 +1,8 @@
 module github.com/jitsucom/bulker/ingest
 
-go 1.22.0
+go 1.23
 
-toolchain go1.22.4
+toolchain go1.23.1
 
 require (
 	github.com/confluentinc/confluent-kafka-go/v2 v2.5.0
@@ -11,6 +11,7 @@ require (
 	github.com/penglongli/gin-metrics v0.1.10
 	github.com/prometheus/client_golang v1.19.1
 	github.com/spf13/viper v1.19.0
+	golang.org/x/net v0.27.0
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1
 )
 
@@ -65,7 +66,6 @@ require (
 	golang.org/x/arch v0.8.0 // indirect
 	golang.org/x/crypto v0.25.0 // indirect
 	golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 // indirect
-	golang.org/x/net v0.27.0 // indirect
 	golang.org/x/oauth2 v0.21.0 // indirect
 	golang.org/x/sys v0.22.0 // indirect
 	golang.org/x/text v0.16.0 // indirect

ingress-manager/go.mod

@@ -1,8 +1,8 @@
 module github.com/jitsucom/bulker/ingress-manager
 
-go 1.22.0
+go 1.23
 
-toolchain go1.22.4
+toolchain go1.23.1
 
 require (
 	cloud.google.com/go/certificatemanager v1.8.3