replace reviewdog action with official golangci action

fix lints

.github/workflows/flow.yml

@@ -38,9 +38,9 @@ jobs:
         run: |
           ./generate_protos.sh
 
-      - uses: actions/setup-go@v4
+      - uses: actions/setup-go@v5
         with:
-          go-version: ">=1.21.0"
+          go-version: "1.21"
           cache-dependency-path: flow/go.sum
 
       - name: install gotestsum

.github/workflows/golang-lint.yml

@@ -5,35 +5,30 @@ on:
     branches: [main]
     paths: [flow/**]
 
+permissions:
+  contents: read
+
 jobs:
-  golangci-lint:
-    permissions:
-      checks: write
-      contents: read
-      pull-requests: write
-    strategy:
-      matrix:
-        runner: [ubicloud-standard-4-ubuntu-2204-arm]
-    runs-on: ${{ matrix.runner }}
+  golangci:
+    name: lint
+    runs-on: [ubicloud-standard-4-ubuntu-2204-arm]
     steps:
-      - name: checkout
-        uses: actions/checkout@v4
-        with:
-          submodules: recursive
-
+      - uses: actions/checkout@v4
       - uses: bufbuild/[email protected]
-
       - name: setup protos
         run: |
           ./generate_protos.sh
-
+      - name: install lib-geos
+        run: |
+          sudo apt-get update
+          sudo apt-get install libgeos-dev
+      - uses: actions/setup-go@v5
+        with:
+          go-version: "1.21"
+          cache: false
       - name: golangci-lint
-        uses: reviewdog/action-golangci-lint@v2
+        uses: golangci/golangci-lint-action@v3
         with:
-          workdir: ./flow
-          reporter: github-pr-review
-          github_token: ${{ secrets.GITHUB_TOKEN }}
-          golangci_lint_flags: "--timeout 10m"
-          fail_on_error: true
-        env:
-          REVIEWDOG_TOKEN: ${{ secrets.REVIEWDOG_TOKEN }}
+          version: v1.55
+          working-directory: ./flow
+          args: --timeout=10m

flow/cmd/mirror_status.go

@@ -103,21 +103,21 @@ func (h *FlowRequestHandler) cloneTableSummary(
 	flowJobName string,
 ) ([]*protos.CloneTableSummary, error) {
 	q := `
-	SELECT 
+	SELECT
 		qp.flow_name,
 		qr.config_proto,
 		MIN(qp.start_time) AS StartTime,
 		COUNT(*) AS NumPartitionsTotal,
 		COUNT(CASE WHEN qp.end_time IS NOT NULL THEN 1 END) AS NumPartitionsCompleted,
 		SUM(qp.rows_in_partition) FILTER (WHERE qp.end_time IS NOT NULL) AS NumRowsSynced,
 		AVG(EXTRACT(EPOCH FROM (qp.end_time - qp.start_time)) * 1000) FILTER (WHERE qp.end_time IS NOT NULL) AS AvgTimePerPartitionMs
-	FROM 
+	FROM
 		peerdb_stats.qrep_partitions qp
 	JOIN
 		peerdb_stats.qrep_runs qr
 	ON
 		qp.flow_name = qr.flow_name
-	WHERE 
+	WHERE
 		qp.flow_name ILIKE $1
 	GROUP BY
 		qp.flow_name, qr.config_proto;
@@ -188,7 +188,6 @@ func (h *FlowRequestHandler) cloneTableSummary(
 		}
 
 		cloneStatuses = append(cloneStatuses, &res)
-
 	}
 	return cloneStatuses, nil
 }

flow/connectors/bigquery/merge_stmt_generator.go

@@ -121,7 +121,7 @@ func (m *mergeStmtGenerator) generateDeDupedCTE() string {
 }
 
 // generateMergeStmt generates a merge statement.
-func (m *mergeStmtGenerator) generateMergeStmt(unchangedToastColumns []string) string {
+func (m *mergeStmtGenerator) generateMergeStmt(_unchangedToastColumns []string) string {
 	// comma separated list of column names
 	columnCount := utils.TableSchemaColumns(m.normalizedTableSchema)
 	backtickColNames := make([]string, 0, columnCount)

flow/connectors/snowflake/snowflake.go

@@ -652,6 +652,9 @@ func (c *SnowflakeConnector) NormalizeRecords(req *model.NormalizeRecordsRequest
 				},
 			}
 			mergeStatement, err := mergeGen.generateMergeStmt()
+			if err != nil {
+				return err
+			}
 
 			startTime := time.Now()
 			c.logger.Info("[merge] merging records...", slog.String("destTable", tableName))