increase reading time and giving Snappy another chance

PeerDB-io · Nov 17, 2023 · aa97878 · aa97878
1 parent f1dd937
commit aa97878
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 2 deletions.
diff --git a/.github/workflows/flow.yml b/.github/workflows/flow.yml
@@ -120,4 +120,4 @@ jobs:
           PEERDB_CATALOG_USER: postgres
           PEERDB_CATALOG_PASSWORD: postgres
           PEERDB_CATALOG_DATABASE: postgres
-          PEERDB_CDC_IDLE_TIMEOUT_SECONDS: 3
+          PEERDB_CDC_IDLE_TIMEOUT_SECONDS: 10
diff --git a/flow/connectors/bigquery/qrep_avro_sync.go b/flow/connectors/bigquery/qrep_avro_sync.go
@@ -328,7 +328,7 @@ func (s *QRepAvroSyncMethod) writeToStage(
 	var avroFilePath string
 	numRecords, err := func() (int, error) {
 		ocfWriter := avro.NewPeerDBOCFWriter(s.connector.ctx, stream, avroSchema,
-			avro.CompressDeflate, qvalue.QDWHTypeBigQuery)
+			avro.CompressSnappy, qvalue.QDWHTypeBigQuery)
 		if s.gcsBucket != "" {
 			bucket := s.connector.storageClient.Bucket(s.gcsBucket)
 			avroFilePath = fmt.Sprintf("%s/%s.avro.snappy", objectFolder, syncID)
@@ -360,6 +360,9 @@ func (s *QRepAvroSyncMethod) writeToStage(
 	if err != nil {
 		return 0, err
 	}
+	if numRecords == 0 {
+		return 0, nil
+	}
 	log.WithFields(log.Fields{
 		"batchOrPartitionID": syncID,
 	}).Infof("wrote %d records to file %s", numRecords, avroFilePath)

diff --git a/flow/e2e/snowflake/peer_flow_sf_test.go b/flow/e2e/snowflake/peer_flow_sf_test.go
@@ -66,6 +66,10 @@ func TestPeerFlowE2ETestSuiteSF(t *testing.T) {
 
 		t.Run(tt.name, tt.test)
 	}
+
+	t.Cleanup(func() {
+		s.TearDownSuite()
+	})
 }
 
 func (s *PeerFlowE2ETestSuiteSF) attachSchemaSuffix(tableName string) string {