Merge branch 'master' into fix/mongodb_type

Author: vr-varad

.github/workflows/performance-test.yml

@@ -95,8 +95,20 @@ jobs:
     - 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; }
+        # Delete all Glue databases matching prefix
+        for db in $(aws glue get-databases \
+          --query "DatabaseList[?starts_with(Name, 'performance_${{ matrix.driver }}')].Name" \
+          --output text); do
+          echo "Deleting Glue database: $db"
+          aws glue delete-database --name "$db" \
+          || echo "Failed to delete Glue database: $db"
+        done
+
+        # Delete corresponding S3 path
+        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

Dockerfile

@@ -12,8 +12,8 @@ RUN go build -o /olake main.go
 # Final Runtime Stage
 FROM alpine:3.18
 
-# Install Java 17 instead of Java 11
-RUN apk add --no-cache openjdk17
+# Install Java 17 and iproute2 for ss command
+RUN apk add --no-cache openjdk17 iproute2
 
 # Copy the binary from the build stage
 COPY --from=base /olake /home/olake

README.md

@@ -72,7 +72,7 @@
 
 | Source        | Full Load    |  CDC          | Incremental       | Notes                       | Documentation               |
 |---------------|--------------|---------------|-------------------|-----------------------------|-----------------------------|
-| PostgreSQL    | ✅           | ✅ `wal2json` | ✅                 |`pgoutput` support WIP       |[Postgres Docs](https://olake.io/docs/connectors/postgres/overview) |
+| PostgreSQL    | ✅           | ✅ `pgoutput` | ✅                 |`wal2json` deprecated       |[Postgres Docs](https://olake.io/docs/connectors/postgres/overview) |
 | MySQL         | ✅           | ✅            | ✅                | Binlog-based CDC            | [MySQL Docs](https://olake.io/docs/connectors/mysql/overview) |
 | MongoDB       | ✅           | ✅            | ✅                | Oplog-based CDC             |[MongoDB Docs](https://olake.io/docs/connectors/mongodb/overview) |
 | Oracle        | ✅           | WIP  | ✅                |  JDBC based Full Load & Incremental                |  [Oracle Docs](https://olake.io/docs/connectors/oracle/overview) |

destination/iceberg/iceberg.go

@@ -17,6 +17,7 @@ import (
 	"github.com/datazip-inc/olake/utils"
 	"github.com/datazip-inc/olake/utils/logger"
 	"github.com/datazip-inc/olake/utils/typeutils"
+	"github.com/spf13/viper"
 )
 
 type Iceberg struct {
@@ -261,8 +262,13 @@ func (i *Iceberg) Check(ctx context.Context) error {
 	i.options = &destination.Options{
 		ThreadID: "test_iceberg_destination",
 	}
+
+	destinationDB := "test_olake"
+	if prefix := viper.GetString(constants.DestinationDatabasePrefix); prefix != "" {
+		destinationDB = fmt.Sprintf("%s_%s", utils.Reformat(prefix), destinationDB)
+	}
 	// Create a temporary setup for checking
-	server, err := newIcebergClient(i.config, []PartitionInfo{}, i.options.ThreadID, true, false, "test_olake")
+	server, err := newIcebergClient(i.config, []PartitionInfo{}, i.options.ThreadID, true, false, destinationDB)
 	if err != nil {
 		return fmt.Errorf("failed to setup iceberg server: %s", err)
 	}
@@ -281,7 +287,7 @@ func (i *Iceberg) Check(ctx context.Context) error {
 		Type: proto.IcebergPayload_GET_OR_CREATE_TABLE,
 		Metadata: &proto.IcebergPayload_Metadata{
 			ThreadId:      server.serverID,
-			DestTableName: "test_olake",
+			DestTableName: destinationDB,
 			Schema:        icebergRawSchema(),
 		},
 	}
@@ -296,7 +302,7 @@ func (i *Iceberg) Check(ctx context.Context) error {
 	// try writing record in dest table
 	currentTime := time.Now().UTC()
 	protoSchema := icebergRawSchema()
-	record := types.CreateRawRecord("olake_test", map[string]any{"name": "olake"}, "r", &currentTime)
+	record := types.CreateRawRecord(destinationDB, map[string]any{"name": "olake"}, "r", &currentTime)
 	protoColumns, err := rawDataColumnBuffer(record, protoSchema)
 	if err != nil {
 		return fmt.Errorf("failed to create raw data column buffer: %s", err)
@@ -305,7 +311,7 @@ func (i *Iceberg) Check(ctx context.Context) error {
 		Type: proto.IcebergPayload_RECORDS,
 		Metadata: &proto.IcebergPayload_Metadata{
 			ThreadId:      server.serverID,
-			DestTableName: "test_olake",
+			DestTableName: destinationDB,
 			Schema:        protoSchema,
 		},
 		Records: []*proto.IcebergPayload_IceRecord{{