chore: staging -> master v0.1.10 (#455)

Co-authored-by: Vishal-datazip <vishal@datazip.io>
Co-authored-by: Ankit Sharma <111491139+hash-data@users.noreply.github.com>
Co-authored-by: hash-data <ankit@datazip.io>

Author: 3 people

.github/workflows/integration-tests.yml

@@ -87,7 +87,7 @@ jobs:
         run: mvn clean package -DskipTests
 
       - name: Run Integration Tests
-        run: go test -v -p 2 ./drivers/mysql/internal/... ./drivers/postgres/internal/... -tags=integration --timeout 0
+        run: go test -v -p 2 ./drivers/mysql/internal/... ./drivers/postgres/internal/... -timeout 0 -run 'Integration'
 
       - name: Cleanup
         if: always()

.github/workflows/performance-test.yml

@@ -0,0 +1,106 @@
+name: Performance Tests
+
+on:
+  push:
+    branches:
+    - "staging"
+    paths:
+    - '**/*.go'
+    - '**/*.java'
+
+permissions:
+  id-token: write
+  contents: read
+
+jobs:
+  performance-tests:
+    environment: Performance Testing
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        include:
+        - driver: mysql
+          driver_upper: MYSQL
+        - driver: postgres
+          driver_upper: POSTGRES
+        # TODO: add benchmark tests for the below databases
+        # - driver: mongodb
+        #   driver_upper: MONGODB
+        # - driver: oracle
+        #   driver_upper: ORACLE
+      fail-fast: false
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v3
+
+    - name: Set up Go
+      uses: actions/setup-go@v4
+      with:
+        go-version: '1.23.2'
+
+    - name: Set up Java for Maven
+      uses: actions/setup-java@v3
+      with:
+        distribution: 'temurin'
+        java-version: '17'
+
+    - name: Setup VPN Client
+      if: matrix.driver != 'oracle'
+      run: |
+        sudo apt-get update -qq && sudo apt-get install -y openvpn iproute2 iputils-ping netcat-openbsd telnet
+        sudo mkdir -p /etc/openvpn/client && sudo chmod 700 /etc/openvpn/client
+        echo "${{ secrets[format('{0}_OPENVPN_CONFIG', matrix.driver_upper)] }}" | base64 --decode | sudo tee /etc/openvpn/client/client.ovpn > /dev/null
+        echo "${{ secrets.OPENVPN_USERNAME }}" | sudo tee /etc/openvpn/client/auth.txt > /dev/null
+        sudo chmod 600 /etc/openvpn/client/client.ovpn /etc/openvpn/client/auth.txt
+        sudo chown root:root /etc/openvpn/client/client.ovpn /etc/openvpn/client/auth.txt
+        sudo openvpn --config /etc/openvpn/client/client.ovpn --daemon ovpn-client --log /var/log/openvpn-client.log --verb 3
+        echo "Establishing VPN connection..."
+        for i in {1..30}; do
+          if ip addr show | grep -q "tun0\|tap0"; then echo "✅ VPN connected"; break; fi
+          [ $i -eq 30 ] && { echo "❌ VPN timeout"; sudo cat /var/log/openvpn-client.log; exit 1; }
+          sleep 2
+        done
+        sudo resolvectl dns tun0 ${{ secrets.VPN_DNS_SERVER }}
+        sudo resolvectl domain tun0 ~private.postgres.database.azure.com
+
+    - name: Configure AWS Credentials
+      uses: aws-actions/configure-aws-credentials@v4
+      with:
+        role-session-name: performance_test_gh
+        role-to-assume: ${{ secrets.AWS_GITHUB_ROLE }}
+        aws-region: ${{ secrets.AWS_REGION }}
+
+    - name: Install Go Dependencies
+      run: go mod download
+
+    - name: Build Project
+      run: go build -v ./...
+
+    - name: Build Iceberg Sink
+      working-directory: ./destination/iceberg/olake-iceberg-java-writer
+      run: mvn clean package -DskipTests
+
+    - name: Create source
+      run: |
+        echo '${{ secrets[format('{0}_SOURCE_JSON', matrix.driver_upper)] }}' | base64 --decode > ./drivers/${{ matrix.driver }}/internal/testdata/source.json
+
+    - name: Create destination
+      run: |
+        echo '${{ secrets[format('{0}_DESTINATION_JSON', matrix.driver_upper)] }}' | base64 --decode > ./drivers/${{ matrix.driver }}/internal/testdata/destination.json
+
+    - name: Run Performance Tests
+      run: go test -v ./drivers/${{ matrix.driver }}/internal/... -timeout 0 -run 'Performance'
+
+    - name: Cleanup
+      if: always()
+      run: |
+        aws glue delete-database --name performance_${{ matrix.driver }} || { echo "failed to delete glue database: performance_${{ matrix.driver }}"; aws glue get-database --name performance_${{ matrix.driver }} || true; }
+        aws s3 rm s3://dz-stag-github-actions/performance_${{ matrix.driver }} --recursive || { echo "failed to delete s3 bucket: performance_${{ matrix.driver }}"; aws s3 ls s3://dz-stag-github-actions/performance_${{ matrix.driver }} || true; }
+        echo "Catalog cleanup completed"
+
+        if [[ "${{ matrix.driver }}" != "oracle" ]]; then
+            sudo pkill -f "openvpn.*client.ovpn" || true
+            sudo rm -rf /etc/openvpn/client/ /var/log/openvpn-client.log || true
+            echo "VPN cleanup completed"
+        fi

destination/iceberg/iceberg_utils.go

@@ -30,10 +30,8 @@ func determineMaxBatchSize() int64 {
 	var batchSize int64
 
 	switch {
-	case ramGB <= 8:
-		batchSize = 100 * 1024 * 1024 // 100MB
 	case ramGB <= 16:
-		batchSize = 200 * 1024 * 1024 // 200MB
+		batchSize = 100 * 1024 * 1024 // 100MB
 	case ramGB <= 32:
 		batchSize = 400 * 1024 * 1024 // 400MB
 	default:

destination/iceberg/olake-iceberg-java-writer/src/main/java/io/debezium/server/iceberg/tableoperator/IcebergTableOperator.java

@@ -160,8 +160,17 @@ private void applyFieldAddition(Table icebergTable, Schema newSchema) {
     // @NOTE avoid committing when there is no schema change. commit creates new
     // commit even when there is no change!
     if (!icebergTable.schema().sameSchema(newSchemaCombined)) {
-      LOGGER.warn("Extending schema of {}", icebergTable.name());
-      us.commit();
+      synchronized (commitLock) {
+        icebergTable.refresh();
+        UpdateSchema usFinal = icebergTable.updateSchema().unionByNameWith(newSchema);
+        if (createIdentifierFields) {
+          usFinal.setIdentifierFields(newSchema.identifierFieldNames());
+        }
+        if (!icebergTable.schema().sameSchema(newSchemaCombined)) {
+          LOGGER.warn("Extending schema of {}", icebergTable.name());
+          usFinal.commit();
+        }
+      }
     }
   }
 

drivers/abstract/abstract.go

@@ -92,13 +92,18 @@ func (a *AbstractDriver) Discover(ctx context.Context) ([]*types.Stream, error)
 		convStream.WithSyncMode(types.FULLREFRESH, types.INCREMENTAL)
 		convStream.SyncMode = types.FULLREFRESH
 
+		// add default columns
+		for column, typ := range DefaultColumns {
+			convStream.UpsertField(column, typ, true)
+		}
+
 		// Add CDC columns if supported
 		if a.driver.CDCSupported() {
-			for column, typ := range DefaultColumns {
-				convStream.UpsertField(column, typ, true)
-			}
 			convStream.WithSyncMode(types.CDC, types.STRICTCDC)
 			convStream.SyncMode = types.CDC
+		} else {
+			// remove cdc column as it is not supported
+			convStream.Schema.Properties.Delete(constants.CdcTimestamp)
 		}
 		finalStreams = append(finalStreams, convStream)
 		return true

drivers/mongodb/go.mod

@@ -81,6 +81,4 @@ require (
 	golang.org/x/text v0.24.0 // indirect
 	golang.org/x/tools v0.30.0 // indirect
 	sigs.k8s.io/yaml v1.4.0 // indirect
-)
-
-replace github.com/datazip-inc/olake => ../../
+)

drivers/mongodb/internal/backfill.go

@@ -71,12 +71,6 @@ func (m *Mongo) GetOrSplitChunks(ctx context.Context, pool *destination.WriterPo
 	logger.Infof("Total expected count for stream %s: %d", stream.ID(), recordCount)
 	pool.AddRecordsToSync(recordCount)
 
-	// build filter
-	filter, err := buildFilter(stream)
-	if err != nil {
-		return nil, fmt.Errorf("failed to parse filter during chunk splitting: %s", err)
-	}
-
 	// check for _id type
 	isObjID, err := isObjectID(ctx, collection)
 	if err != nil {
@@ -88,7 +82,7 @@ func (m *Mongo) GetOrSplitChunks(ctx context.Context, pool *destination.WriterPo
 	var retryErr error
 	var chunksArray []types.Chunk
 	err = abstract.RetryOnBackoff(m.config.RetryCount, 1*time.Minute, func() error {
-		chunksArray, retryErr = m.splitChunks(ctx, collection, stream, filter, isObjID, storageSize)
+		chunksArray, retryErr = m.splitChunks(ctx, collection, stream, isObjID, storageSize)
 		return retryErr
 	})
 	if err != nil {
@@ -97,11 +91,11 @@ func (m *Mongo) GetOrSplitChunks(ctx context.Context, pool *destination.WriterPo
 	return types.NewSet(chunksArray...), nil
 }
 
-func (m *Mongo) splitChunks(ctx context.Context, collection *mongo.Collection, stream types.StreamInterface, filter bson.D, isObjID bool, storageSize float64) ([]types.Chunk, error) {
+func (m *Mongo) splitChunks(ctx context.Context, collection *mongo.Collection, stream types.StreamInterface, isObjID bool, storageSize float64) ([]types.Chunk, error) {
 	splitVectorStrategy := func() ([]types.Chunk, error) {
 		getID := func(order int) (primitive.ObjectID, error) {
 			var doc bson.M
-			err := collection.FindOne(ctx, filter, options.FindOne().SetSort(bson.D{{Key: "_id", Value: order}})).Decode(&doc)
+			err := collection.FindOne(ctx, bson.D{}, options.FindOne().SetSort(bson.D{{Key: "_id", Value: order}})).Decode(&doc)
 			if err == mongo.ErrNoDocuments {
 				return primitive.NilObjectID, nil
 			}
@@ -131,9 +125,6 @@ func (m *Mongo) splitChunks(ctx context.Context, collection *mongo.Collection, s
 				{Key: "maxChunkSize", Value: 1024},
 			}
 
-			if len(filter) > 0 {
-				cmd = append(cmd, bson.E{Key: "filter", Value: filter})
-			}
 			if err := collection.Database().RunCommand(ctx, cmd).Decode(&result); err != nil {
 				return nil, fmt.Errorf("failed to run splitVector command: %s", err)
 			}
@@ -169,19 +160,14 @@ func (m *Mongo) splitChunks(ctx context.Context, collection *mongo.Collection, s
 	bucketAutoStrategy := func(storageSize float64) ([]types.Chunk, error) {
 		logger.Infof("using bucket auto strategy for stream: %s", stream.ID())
 		// Use $bucketAuto for chunking
-		pipeline := mongo.Pipeline{}
-		if len(filter) > 0 {
-			pipeline = append(pipeline, bson.D{{Key: "$match", Value: filter}})
-		}
-
 		numberOfBuckets := int(math.Ceil(storageSize / float64(constants.EffectiveParquetSize)))
-		pipeline = append(pipeline,
-			bson.D{{Key: "$sort", Value: bson.D{{Key: "_id", Value: 1}}}},
-			bson.D{{Key: "$bucketAuto", Value: bson.D{
+		pipeline := mongo.Pipeline{
+			{{Key: "$sort", Value: bson.D{{Key: "_id", Value: 1}}}},
+			{{Key: "$bucketAuto", Value: bson.D{
 				{Key: "groupBy", Value: "$_id"},
 				{Key: "buckets", Value: numberOfBuckets},
 			}}},
-		)
+		}
 
 		cursor, err := collection.Aggregate(ctx, pipeline)
 		if err != nil {
@@ -227,7 +213,7 @@ func (m *Mongo) splitChunks(ctx context.Context, collection *mongo.Collection, s
 
 	timestampStrategy := func() ([]types.Chunk, error) {
 		// Time-based strategy implementation
-		first, last, err := m.fetchExtremes(ctx, collection, filter)
+		first, last, err := m.fetchExtremes(ctx, collection)
 		if err != nil {
 			return nil, err
 		}
@@ -307,12 +293,12 @@ func (m *Mongo) totalCountAndStorageSizeInCollection(ctx context.Context, collec
 	return count, storageSize, nil
 }
 
-func (m *Mongo) fetchExtremes(ctx context.Context, collection *mongo.Collection, filter bson.D) (time.Time, time.Time, error) {
+func (m *Mongo) fetchExtremes(ctx context.Context, collection *mongo.Collection) (time.Time, time.Time, error) {
 	extreme := func(sortby int) (time.Time, error) {
 		// Find the first document
 		var result bson.M
 		// Sort by _id ascending to get the first document
-		err := collection.FindOne(ctx, filter, options.FindOne().SetSort(bson.D{{Key: "_id", Value: sortby}})).Decode(&result)
+		err := collection.FindOne(ctx, bson.D{}, options.FindOne().SetSort(bson.D{{Key: "_id", Value: sortby}})).Decode(&result)
 		if err != nil {
 			return time.Time{}, err
 		}

drivers/mongodb/internal/mon_test.go

@@ -0,0 +1,20 @@
+package driver
+
+import (
+	"testing"
+
+	"github.com/datazip-inc/olake/constants"
+	"github.com/datazip-inc/olake/utils/testutils"
+)
+
+func TestMongodbPerformance(t *testing.T) {
+	config := &testutils.PerformanceTest{
+		TestConfig:      testutils.GetTestConfig(string(constants.MongoDB)),
+		Namespace:       "twitter_data",
+		BackfillStreams: []string{"tweets"},
+		CDCStreams:      []string{"tweets_cdc"},
+		ExecuteQuery:    ExecuteQuery,
+	}
+
+	config.TestPerformance(t)
+}

drivers/mongodb/internal/mon_test_util.go

@@ -0,0 +1,86 @@
+package driver
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/datazip-inc/olake/utils"
+	"github.com/datazip-inc/olake/utils/testutils"
+	"github.com/stretchr/testify/require"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+func ExecuteQuery(ctx context.Context, t *testing.T, streams []string, operation string, fileConfig bool) {
+	t.Helper()
+
+	var connStr string
+	var config Config
+	if fileConfig {
+		utils.UnmarshalFile("./testdata/source.json", &config, false)
+		connStr = fmt.Sprintf(
+			"mongodb://%s:%s@%s/?authSource=%s&readPreference=%s",
+			config.Username,
+			config.Password,
+			strings.Join(config.Hosts, ","),
+			config.AuthDB,
+			config.ReadPreference,
+		)
+	} else {
+		connStr = "mongodb://localhost:27017"
+	}
+	db, ok := mongo.Connect(ctx, options.Client().ApplyURI(connStr))
+	require.NoError(t, ok, "failed to connect to mongodb")
+
+	switch operation {
+	case "setup_cdc":
+		// truncate the cdc tables
+		for _, cdcStream := range streams {
+			_, err := db.Database(config.Database).Collection(cdcStream).DeleteMany(ctx, bson.D{})
+			require.NoError(t, err, fmt.Sprintf("failed to execute %s operation", operation), err)
+		}
+		return
+
+	case "bulk_cdc_data_insert":
+		backfillStreams := testutils.GetBackfillStreamsFromCDC(streams)
+		totalRows := 15000000
+
+		// TODO: insert data in batch
+		// insert the data into the cdc tables concurrently
+		err := utils.Concurrent(ctx, streams, len(streams), func(ctx context.Context, cdcStream string, executionNumber int) error {
+			srcColl := db.Database(config.Database).Collection(backfillStreams[executionNumber-1])
+			destColl := db.Database(config.Database).Collection(cdcStream)
+
+			cursor, err := srcColl.Find(ctx, bson.D{}, options.Find().SetLimit(int64(totalRows)))
+			if err != nil {
+				return fmt.Errorf("stream: %s, error: %s", cdcStream, err)
+			}
+			defer cursor.Close(ctx)
+
+			var docs []interface{}
+			for cursor.Next(ctx) {
+				var doc bson.M
+				if err := cursor.Decode(&doc); err != nil {
+					return err
+				}
+				docs = append(docs, doc)
+			}
+			if err := cursor.Err(); err != nil {
+				return err
+			}
+			if len(docs) == 0 {
+				return nil
+			}
+			_, err = destColl.InsertMany(ctx, docs)
+			if err != nil {
+				return fmt.Errorf("stream: %s, error: %s", cdcStream, err)
+			}
+			return nil
+		})
+		require.NoError(t, err, fmt.Sprintf("failed to execute %s operation", operation), err)
+		return
+	}
+}