Merge branch 'main' into ui/improvements-1

PeerDB-io · Dec 5, 2023 · 8aa9548 · 8aa9548
2 parents bbf3212 + 20d901c
commit 8aa9548
Show file tree

Hide file tree

Showing 18 changed files with 815 additions and 188 deletions.
diff --git a/dev-peerdb.sh b/dev-peerdb.sh
@@ -7,5 +7,6 @@ then
     exit 1
 fi
 
+export PEERDB_VERSION_SHA_SHORT=local-$(git rev-parse --short HEAD)
 docker compose -f docker-compose-dev.yml up --build \
  --no-attach temporal --no-attach pyroscope --no-attach temporal-ui
diff --git a/docker-bake.hcl b/docker-bake.hcl
@@ -28,6 +28,9 @@ target "flow-api" {
     "linux/amd64",
     "linux/arm64",
   ]
+  args = {
+    PEERDB_VERSION_SHA_SHORT = "${SHA_SHORT}"
+  }
   tags = [
     "${REGISTRY}/flow-api:${TAG}",
     "${REGISTRY}/flow-api:${SHA_SHORT}",

diff --git a/docker-compose-dev.yml b/docker-compose-dev.yml
@@ -8,7 +8,7 @@ x-catalog-config: &catalog-config
   PEERDB_CATALOG_DATABASE: postgres
 
 x-flow-worker-env: &flow-worker-env
-  # For Temporal Cloud, this will look like: 
+  # For Temporal Cloud, this will look like:
   # <yournamespace>.<id>.tmprl.cloud:7233
   TEMPORAL_HOST_PORT: temporal:7233
   PEERDB_TEMPORAL_NAMESPACE: default
@@ -115,6 +115,8 @@ services:
       context: .
       dockerfile: stacks/flow.Dockerfile
       target: flow-api
+      args:
+        PEERDB_VERSION_SHA_SHORT: ${PEERDB_VERSION_SHA_SHORT:-}
     ports:
       - 8112:8112
       - 8113:8113

diff --git a/flow/activities/flowable.go b/flow/activities/flowable.go
@@ -764,12 +764,6 @@ func (a *FlowableActivity) QRepWaitUntilNewRows(ctx context.Context,
 		waitBetweenBatches = time.Duration(config.WaitBetweenBatchesSeconds) * time.Second
 	}
 
-	if config.WatermarkColumn == "xmin" {
-		// for xmin we ignore the wait between batches, as seq scan time is
-		// extremely slow.
-		waitBetweenBatches = 10 * time.Second
-	}
-
 	srcConn, err := connectors.GetQRepPullConnector(ctx, config.SourcePeer)
 	if err != nil {
 		return fmt.Errorf("failed to get qrep source connector: %w", err)

diff --git a/flow/connectors/postgres/qrep.go b/flow/connectors/postgres/qrep.go
@@ -274,11 +274,6 @@ func (c *PostgresConnector) getMinMaxValues(
 
 func (c *PostgresConnector) CheckForUpdatedMaxValue(config *protos.QRepConfig,
 	last *protos.QRepPartition) (bool, error) {
-	// for xmin lets always assume there are updates
-	if config.WatermarkColumn == "xmin" {
-		return true, nil
-	}
-
 	tx, err := c.pool.Begin(c.ctx)
 	if err != nil {
 		return false, fmt.Errorf("unable to begin transaction for getting max value: %w", err)
@@ -571,9 +566,9 @@ func (c *PostgresConnector) PullXminRecordStream(
 
 	var numRecords int
 	if partition.Range != nil {
-		numRecords, currentSnapshotXmin, err = executor.ExecuteAndProcessQueryStreamGettingCurrentTxid(stream, query, oldxid)
+		numRecords, currentSnapshotXmin, err = executor.ExecuteAndProcessQueryStreamGettingCurrentSnapshotXmin(stream, query, oldxid)
 	} else {
-		numRecords, currentSnapshotXmin, err = executor.ExecuteAndProcessQueryStreamGettingCurrentTxid(stream, query)
+		numRecords, currentSnapshotXmin, err = executor.ExecuteAndProcessQueryStreamGettingCurrentSnapshotXmin(stream, query)
 	}
 	if err != nil {
 		return 0, currentSnapshotXmin, err

diff --git a/flow/connectors/postgres/qrep_query_executor.go b/flow/connectors/postgres/qrep_query_executor.go
@@ -325,7 +325,7 @@ func (qe *QRepQueryExecutor) ExecuteAndProcessQueryStream(
 	return totalRecordsFetched, err
 }
 
-func (qe *QRepQueryExecutor) ExecuteAndProcessQueryStreamGettingCurrentTxid(
+func (qe *QRepQueryExecutor) ExecuteAndProcessQueryStreamGettingCurrentSnapshotXmin(
 	stream *model.QRecordStream,
 	query string,
 	args ...interface{},

diff --git a/flow/e2e/snowflake/qrep_flow_sf_test.go b/flow/e2e/snowflake/qrep_flow_sf_test.go
@@ -208,7 +208,7 @@ func (s *PeerFlowE2ETestSuiteSF) Test_Complete_QRep_Flow_Avro_SF_Upsert_XMIN() {
 	qrepConfig.WatermarkColumn = "xmin"
 	s.NoError(err)
 
-	e2e.RunQrepFlowWorkflow(env, qrepConfig)
+	e2e.RunXminFlowWorkflow(env, qrepConfig)
 
 	// Verify workflow completes without error
 	s.True(env.IsWorkflowCompleted())

diff --git a/flow/e2e/test_utils.go b/flow/e2e/test_utils.go
@@ -20,7 +20,7 @@ import (
 	"go.temporal.io/sdk/testsuite"
 )
 
-// readFileToBytes reads a file to a byte array.
+// ReadFileToBytes reads a file to a byte array.
 func ReadFileToBytes(path string) ([]byte, error) {
 	var ret []byte
 
@@ -49,6 +49,7 @@ func RegisterWorkflowsAndActivities(env *testsuite.TestWorkflowEnvironment) {
 	env.RegisterWorkflow(peerflow.SnapshotFlowWorkflow)
 	env.RegisterWorkflow(peerflow.NormalizeFlowWorkflow)
 	env.RegisterWorkflow(peerflow.QRepFlowWorkflow)
+	env.RegisterWorkflow(peerflow.XminFlowWorkflow)
 	env.RegisterWorkflow(peerflow.QRepPartitionWorkflow)
 	env.RegisterActivity(&activities.FlowableActivity{})
 	env.RegisterActivity(&activities.SnapshotActivity{})
@@ -303,6 +304,13 @@ func RunQrepFlowWorkflow(env *testsuite.TestWorkflowEnvironment, config *protos.
 	env.ExecuteWorkflow(peerflow.QRepFlowWorkflow, config, state)
 }
 
+func RunXminFlowWorkflow(env *testsuite.TestWorkflowEnvironment, config *protos.QRepConfig) {
+	state := peerflow.NewQRepFlowState()
+	state.LastPartition.PartitionId = uuid.New().String()
+	time.Sleep(5 * time.Second)
+	env.ExecuteWorkflow(peerflow.XminFlowWorkflow, config, state)
+}
+
 func GetOwnersSchema() *model.QRecordSchema {
 	return &model.QRecordSchema{
 		Fields: []*model.QField{