UI for BigQuery Peer (#620)

<img width="944" alt="Screenshot 2023-11-07 at 5 05 55 PM"
src="https://github.com/PeerDB-io/peerdb/assets/65964360/5f0057ad-4544-412f-b14e-315c4bf8e638">

- Adds logo in Select Data source dropdown
- Removes snapshot table since they don't seem to be in flows table
- Disables button and shows message for non-pg source selection in CDC
UI
- Fixes table mapping component
- Adds connection check step in bigquery new connector method in Flow

flow/cmd/handler.go

@@ -500,6 +500,13 @@ func (h *FlowRequestHandler) CreatePeer(
 		}
 		sfConfig := sfConfigObject.SnowflakeConfig
 		encodedConfig, encodingErr = proto.Marshal(sfConfig)
+	case protos.DBType_BIGQUERY:
+		bqConfigObject, ok := config.(*protos.Peer_BigqueryConfig)
+		if !ok {
+			return wrongConfigResponse, nil
+		}
+		bqConfig := bqConfigObject.BigqueryConfig
+		encodedConfig, encodingErr = proto.Marshal(bqConfig)
 	case protos.DBType_SQLSERVER:
 		sqlServerConfigObject, ok := config.(*protos.Peer_SqlserverConfig)
 		if !ok {

flow/connectors/bigquery/bigquery.go

@@ -165,6 +165,13 @@ func NewBigQueryConnector(ctx context.Context, config *protos.BigqueryConfig) (*
 		return nil, fmt.Errorf("failed to create BigQuery client: %v", err)
 	}
 
+	datasetID := config.GetDatasetId()
+	_, checkErr := client.Dataset(datasetID).Metadata(ctx)
+	if checkErr != nil {
+		log.Errorf("failed to get dataset metadata: %v", checkErr)
+		return nil, fmt.Errorf("failed to get dataset metadata: %v", checkErr)
+	}
+
 	storageClient, err := bqsa.CreateStorageClient(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create Storage client: %v", err)
@@ -174,7 +181,7 @@ func NewBigQueryConnector(ctx context.Context, config *protos.BigqueryConfig) (*
 		ctx:           ctx,
 		bqConfig:      config,
 		client:        client,
-		datasetID:     config.GetDatasetId(),
+		datasetID:     datasetID,
 		storageClient: storageClient,
 	}, nil
 }

ui/app/api/peers/route.ts

@@ -46,6 +46,12 @@ const constructPeer = (
         type: DBType.SNOWFLAKE,
         snowflakeConfig: config as SnowflakeConfig,
       };
+    case 'BIGQUERY':
+      return {
+        name,
+        type: DBType.BIGQUERY,
+        bigqueryConfig: config as BigqueryConfig,
+      };
     default:
       return;
   }
@@ -76,6 +82,7 @@ export async function POST(request: Request) {
     return new Response(JSON.stringify(response));
   } else if (mode === 'create') {
     const req: CreatePeerRequest = { peer };
+    console.log('/peer/create req:', req);
     const createStatus: CreatePeerResponse = await fetch(
       `${flowServiceAddr}/v1/peers/create`,
       {

ui/app/dto/PeersDTO.ts

@@ -1,4 +1,8 @@
-import { PostgresConfig, SnowflakeConfig } from '@/grpc_generated/peers';
+import {
+  BigqueryConfig,
+  PostgresConfig,
+  SnowflakeConfig,
+} from '@/grpc_generated/peers';
 
 export type UValidatePeerResponse = {
   valid: boolean;
@@ -27,7 +31,7 @@ export type UDropPeerResponse = {
   errorMessage: string;
 };
 
-export type PeerConfig = PostgresConfig | SnowflakeConfig;
+export type PeerConfig = PostgresConfig | SnowflakeConfig | BigqueryConfig;
 export type CatalogPeer = {
   id: number;
   name: string;