fixed issue, and BigQuery compression is dead

PeerDB-io · Nov 22, 2023 · abb795e · abb795e
1 parent e93258c
commit abb795e
Show file tree

Hide file tree

Showing 4 changed files with 7 additions and 7 deletions.
diff --git a/flow/connectors/bigquery/qrep_avro_sync.go b/flow/connectors/bigquery/qrep_avro_sync.go
@@ -331,7 +331,7 @@ func (s *QRepAvroSyncMethod) writeToStage(
 			avro.CompressNone, qvalue.QDWHTypeBigQuery)
 		if s.gcsBucket != "" {
 			bucket := s.connector.storageClient.Bucket(s.gcsBucket)
-			avroFilePath = fmt.Sprintf("%s/%s.avro.snappy", objectFolder, syncID)
+			avroFilePath = fmt.Sprintf("%s/%s.avro", objectFolder, syncID)
 			obj := bucket.Object(avroFilePath)
 			w := obj.NewWriter(s.connector.ctx)
 
@@ -346,7 +346,7 @@ func (s *QRepAvroSyncMethod) writeToStage(
 				return 0, fmt.Errorf("failed to create temp dir: %w", err)
 			}
 
-			avroFilePath = fmt.Sprintf("%s/%s.avro.snappy", tmpDir, syncID)
+			avroFilePath = fmt.Sprintf("%s/%s.avro", tmpDir, syncID)
 			log.WithFields(log.Fields{
 				"batchOrPartitionID": syncID,
 			}).Infof("writing records to local file %s", avroFilePath)

diff --git a/flow/e2e/bigquery/peer_flow_bq_test.go b/flow/e2e/bigquery/peer_flow_bq_test.go
@@ -253,7 +253,7 @@ func (s *PeerFlowE2ETestSuiteBQ) Test_Complete_Simple_Flow_BQ() {
 	s.NoError(err)
 
 	limits := peerflow.CDCFlowLimits{
-		TotalSyncFlows: 3,
+		TotalSyncFlows: 2,
 		MaxBatchSize:   100,
 	}
 

diff --git a/flow/e2e/bigquery/qrep_flow_bq_test.go b/flow/e2e/bigquery/qrep_flow_bq_test.go
@@ -39,9 +39,9 @@ func (s *PeerFlowE2ETestSuiteBQ) compareTableContentsBQ(tableName string, colsSt
 
 	// read rows from destination table
 	qualifiedTableName := fmt.Sprintf("`%s.%s`", s.bqHelper.Config.DatasetId, tableName)
-	bqRows, err := s.bqHelper.ExecuteAndProcessQuery(
-		fmt.Sprintf("SELECT %s FROM %s ORDER BY id", colsString, qualifiedTableName),
-	)
+	bqSelQuery := fmt.Sprintf("SELECT %s FROM %s ORDER BY id", colsString, qualifiedTableName)
+	fmt.Printf("running query on bigquery: %s\n", bqSelQuery)
+	bqRows, err := s.bqHelper.ExecuteAndProcessQuery(bqSelQuery)
 	s.NoError(err)
 
 	s.True(pgRows.Equals(bqRows), "rows from source and destination tables are not equal")

diff --git a/flow/e2e/test_utils.go b/flow/e2e/test_utils.go
@@ -71,7 +71,7 @@ func SetupCDCFlowStatusQuery(env *testsuite.TestWorkflowEnvironment,
 				log.Errorln(err)
 			}
 
-			if state.SetupComplete {
+			if state.SnapshotComplete {
 				break
 			}
 		} else {