Merge branch 'main' into fix-writemode

PeerDB-io · Oct 5, 2023 · a7eea0f · a7eea0f
2 parents 4c13df4 + 5b9f98d
commit a7eea0f
Show file tree

Hide file tree

Showing 30 changed files with 1,033 additions and 283 deletions.
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -4,7 +4,9 @@ on:
   push:
     branches: [main, release/*]
   pull_request:
-    branches: [main]
+    branches:
+      - "main"
+      - "release/*"
 
 jobs:
   build:
@@ -31,7 +33,6 @@ jobs:
       - uses: actions/checkout@v3
         with:
           submodules: recursive
-          token: ${{ secrets.SUBMODULE_CHECKOUT }}
 
       - uses: dtolnay/rust-toolchain@stable
 

diff --git a/.github/workflows/dev-debian.yml b/.github/workflows/dev-debian.yml
diff --git a/.github/workflows/dev-docker.yml b/.github/workflows/dev-docker.yml
@@ -3,14 +3,12 @@ name: Dev Docker Images
 on:
   push:
     branches: [main]
-  pull_request:
-    branches: [main]
 
 jobs:
   docker-build:
     strategy:
       matrix:
-        runner: [ubuntu-latest, ubicloud]
+        runner: [ubicloud]
     runs-on: ${{ matrix.runner }}
     permissions:
       contents: read
@@ -20,7 +18,6 @@ jobs:
         uses: actions/checkout@v3
         with:
           submodules: recursive
-          token: ${{ secrets.SUBMODULE_CHECKOUT }}
 
       - uses: depot/setup-action@v1
 

diff --git a/.github/workflows/flow.yml b/.github/workflows/flow.yml
@@ -2,7 +2,8 @@ name: Flow build and test
 
 on:
   pull_request:
-    branches: [main]
+    branches:
+      - "main"
   push:
     branches: [main]
 
@@ -34,7 +35,7 @@ jobs:
 
       - uses: actions/setup-go@v3
         with:
-          go-version: '>=1.19.0'
+          go-version: ">=1.19.0"
 
       - name: install gotestsum
         run: |
@@ -73,7 +74,7 @@ jobs:
 
       - name: run tests
         run: |
-          gotestsum --format testname -- -p 4 ./... -timeout 1200s
+          gotestsum --format testname -- -p 4 ./... -timeout 2400s
         working-directory: ./flow
         env:
           AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}

diff --git a/.github/workflows/golang-lint.yml b/.github/workflows/golang-lint.yml
@@ -1,6 +1,9 @@
 name: GolangCI-Lint
 
-on: [pull_request]
+on:
+  pull_request:
+    branches:
+      - "main"
 
 jobs:
   golangci-lint:
@@ -10,14 +13,13 @@ jobs:
       pull-requests: write
     strategy:
       matrix:
-        runner: [ubuntu-latest, ubicloud]
+        runner: [ubicloud]
     runs-on: ${{ matrix.runner }}
     steps:
       - name: checkout
         uses: actions/checkout@v3
         with:
           submodules: recursive
-          token: ${{ secrets.SUBMODULE_CHECKOUT }}
 
       - name: golangci-lint
         uses: reviewdog/action-golangci-lint@v2

diff --git a/.github/workflows/rust-lint.yml b/.github/workflows/rust-lint.yml
@@ -1,5 +1,10 @@
 name: clippy-action
-on: [pull_request]
+
+on:
+  pull_request:
+    branches:
+      - "main"
+
 jobs:
   clippy:
     permissions:
@@ -8,23 +13,18 @@ jobs:
       pull-requests: write
     strategy:
       matrix:
-        runner: [ubuntu-latest, ubicloud]
+        runner: [ubicloud-standard-4]
     runs-on: ${{ matrix.runner }}
     steps:
-    - name: checkout
-      uses: actions/checkout@v3
-      with:
-        submodules: recursive
-        token: ${{ secrets.SUBMODULE_CHECKOUT }}
+      - name: checkout
+        uses: actions/checkout@v3
+        with:
+          submodules: recursive
 
-    - uses: dtolnay/rust-toolchain@stable
-      with:
-        components: clippy
+      - uses: dtolnay/rust-toolchain@stable
+        with:
+          components: clippy
 
-    - uses: giraffate/clippy-action@v1
-      with:
-        reporter: 'github-pr-review'
-        github_token: ${{ secrets.GITHUB_TOKEN }}
-        workdir: ./nexus
-      env:
-        REVIEWDOG_TOKEN: ${{ secrets.REVIEWDOG_TOKEN }}
+      - name: clippy
+        run: cargo clippy -- -D warnings
+        working-directory: ./nexus
diff --git a/.github/workflows/stable-debian.yml b/.github/workflows/stable-debian.yml
diff --git a/.github/workflows/stable-docker.yml b/.github/workflows/stable-docker.yml
@@ -9,7 +9,7 @@ jobs:
   docker-build:
     strategy:
       matrix:
-        runner: [ubuntu-latest, ubicloud]
+        runner: [ubicloud]
     runs-on: ${{ matrix.runner }}
     permissions:
       contents: read
@@ -19,7 +19,6 @@ jobs:
         uses: actions/checkout@v3
         with:
           submodules: recursive
-          token: ${{ secrets.SUBMODULE_CHECKOUT }}
 
       - uses: depot/setup-action@v1
 

diff --git a/.github/workflows/ui-build.yml b/.github/workflows/ui-build.yml
@@ -13,7 +13,7 @@ jobs:
     name: Build & Test UI
     strategy:
       matrix:
-        runner: [ubuntu-latest, ubicloud]
+        runner: [ubicloud]
     runs-on: ${{ matrix.runner }}
     steps:
       - name: checkout
@@ -25,4 +25,4 @@ jobs:
 
       - name: Build
         working-directory: ui
-        run: yarn build
+        run: yarn build
diff --git a/.github/workflows/ui-lint.yml b/.github/workflows/ui-lint.yml
@@ -17,7 +17,7 @@ jobs:
     name: Run UI linters
     strategy:
       matrix:
-        runner: [ubuntu-latest, ubicloud]
+        runner: [ubicloud]
     runs-on: ${{ matrix.runner }}
     steps:
       - name: checkout
@@ -33,4 +33,4 @@ jobs:
           eslint: true
           prettier: true
           eslint_dir: ui
-          prettier_dir: ui
+          prettier_dir: ui
diff --git a/flow/connectors/bigquery/bigquery.go b/flow/connectors/bigquery/bigquery.go
@@ -214,7 +214,7 @@ func (c *BigQueryConnector) ReplayTableSchemaDelta(flowJobName string,
 	}
 
 	for _, droppedColumn := range schemaDelta.DroppedColumns {
-		_, err := c.client.Query(fmt.Sprintf("ALTER TABLE %s.%s DROP COLUMN %s", c.datasetID,
+		_, err := c.client.Query(fmt.Sprintf("ALTER TABLE %s.%s DROP COLUMN `%s`", c.datasetID,
 			schemaDelta.DstTableName, droppedColumn)).Read(c.ctx)
 		if err != nil {
 			return fmt.Errorf("failed to drop column %s for table %s: %w", droppedColumn,
@@ -226,8 +226,9 @@ func (c *BigQueryConnector) ReplayTableSchemaDelta(flowJobName string,
 		}).Infof("[schema delta replay] dropped column %s", droppedColumn)
 	}
 	for _, addedColumn := range schemaDelta.AddedColumns {
-		_, err := c.client.Query(fmt.Sprintf("ALTER TABLE %s.%s ADD COLUMN %s %s", c.datasetID,
-			schemaDelta.DstTableName, addedColumn.ColumnName, addedColumn.ColumnType)).Read(c.ctx)
+		_, err := c.client.Query(fmt.Sprintf("ALTER TABLE %s.%s ADD COLUMN `%s` %s", c.datasetID,
+			schemaDelta.DstTableName, addedColumn.ColumnName,
+			qValueKindToBigQueryType(addedColumn.ColumnType))).Read(c.ctx)
 		if err != nil {
 			return fmt.Errorf("failed to add column %s for table %s: %w", addedColumn.ColumnName,
 				schemaDelta.SrcTableName, err)

diff --git a/flow/connectors/postgres/cdc.go b/flow/connectors/postgres/cdc.go
@@ -545,13 +545,17 @@ func (p *PostgresCDCSource) processRelationMessage(
 	// retrieve initial RelationMessage for table changed.
 	prevRel := p.relationMessageMapping[currRel.RelationId]
 	// creating maps for lookup later
-	prevRelMap := make(map[string]bool)
-	currRelMap := make(map[string]bool)
+	prevRelMap := make(map[string]*protos.PostgresTableIdentifier)
+	currRelMap := make(map[string]*protos.PostgresTableIdentifier)
 	for _, column := range prevRel.Columns {
-		prevRelMap[column.Name] = true
+		prevRelMap[column.Name] = &protos.PostgresTableIdentifier{
+			RelId: column.DataType,
+		}
 	}
 	for _, column := range currRel.Columns {
-		currRelMap[column.Name] = true
+		currRelMap[column.Name] = &protos.PostgresTableIdentifier{
+			RelId: column.DataType,
+		}
 	}
 
 	schemaDelta := &protos.TableSchemaDelta{
@@ -564,7 +568,15 @@ func (p *PostgresCDCSource) processRelationMessage(
 	}
 	for _, column := range currRel.Columns {
 		// not present in previous relation message, but in current one, so added.
-		if !prevRelMap[column.Name] {
+		if prevRelMap[column.Name] == nil {
+			schemaDelta.AddedColumns = append(schemaDelta.AddedColumns, &protos.DeltaAddedColumn{
+				ColumnName: column.Name,
+				ColumnType: string(postgresOIDToQValueKind(column.DataType)),
+			})
+			// present in previous and current relation messages, but data types have changed.
+			// so we add it to AddedColumns and DroppedColumns, knowing that we process DroppedColumns first.
+		} else if prevRelMap[column.Name].RelId != currRelMap[column.Name].RelId {
+			schemaDelta.DroppedColumns = append(schemaDelta.DroppedColumns, column.Name)
 			schemaDelta.AddedColumns = append(schemaDelta.AddedColumns, &protos.DeltaAddedColumn{
 				ColumnName: column.Name,
 				ColumnType: string(postgresOIDToQValueKind(column.DataType)),
@@ -573,7 +585,7 @@ func (p *PostgresCDCSource) processRelationMessage(
 	}
 	for _, column := range prevRel.Columns {
 		// present in previous relation message, but not in current one, so dropped.
-		if !currRelMap[column.Name] {
+		if currRelMap[column.Name] == nil {
 			schemaDelta.DroppedColumns = append(schemaDelta.DroppedColumns, column.Name)
 		}
 	}

diff --git a/flow/connectors/postgres/client.go b/flow/connectors/postgres/client.go
@@ -516,10 +516,10 @@ func (c *PostgresConnector) generateMergeStatement(destinationTableIdentifier st
 		pgType := qValueKindToPostgresType(genericColumnType)
 		if strings.Contains(genericColumnType, "array") {
 			flattenedCastsSQLArray = append(flattenedCastsSQLArray,
-				fmt.Sprintf("ARRAY(SELECT * FROM JSON_ARRAY_ELEMENTS_TEXT((_peerdb_data->>'%s')::JSON))::%s AS %s",
+				fmt.Sprintf("ARRAY(SELECT * FROM JSON_ARRAY_ELEMENTS_TEXT((_peerdb_data->>'%s')::JSON))::%s AS \"%s\"",
 					strings.Trim(columnName, "\""), pgType, columnName))
 		} else {
-			flattenedCastsSQLArray = append(flattenedCastsSQLArray, fmt.Sprintf("(_peerdb_data->>'%s')::%s AS %s",
+			flattenedCastsSQLArray = append(flattenedCastsSQLArray, fmt.Sprintf("(_peerdb_data->>'%s')::%s AS \"%s\"",
 				strings.Trim(columnName, "\""), pgType, columnName))
 		}
 		if normalizedTableSchema.PrimaryKeyColumn == columnName {