Method TestTransform

flow/e2e/postgres_qrep_test.go:497–545

()

}

func (s PeerFlowE2ETestSuitePG) TestTransform() {
	numRows := 10

	srcTable := "test_transform"
	s.setupSourceTable(srcTable, numRows)

	dstTable := "test_transformdst"

	srcSchemaQualified := fmt.Sprintf("%s_%s.%s", "e2e_test", s.suffix, srcTable)
	dstSchemaQualified := fmt.Sprintf("%s_%s.%s", "e2e_test", s.suffix, dstTable)

	query := fmt.Sprintf("SELECT * FROM %s WHERE updated_at BETWEEN {{.start}} AND {{.end}}", srcSchemaQualified)

	_, err := s.Conn().Exec(s.t.Context(), `insert into public.scripts (name, lang, source) values
	('pgtransform', 'lua', 'function transformRow(row) row.myreal = 1729 end') on conflict do nothing`)
	require.NoError(s.t, err)

	jobName := AddSuffix(s, srcTable)
	qrepConfig := CreateQRepWorkflowConfig(
		s.t,
		jobName,
		srcSchemaQualified,
		dstSchemaQualified,
		query,
		GeneratePostgresPeer(s.t).Name,
		"",
		true,
		"_PEERDB_SYNCED_AT",
		"",
	)
	qrepConfig.WriteMode = &protos.QRepWriteMode{
		WriteType: protos.QRepWriteType_QREP_WRITE_MODE_OVERWRITE,
	}
	qrepConfig.InitialCopyOnly = false
	qrepConfig.Script = "pgtransform"

	tc := NewTemporalClient(s.t)
	env := RunQRepFlowWorkflow(s.t, tc, qrepConfig)
	EnvWaitFor(s.t, env, 3*time.Minute, "waiting for first sync to complete", func() bool {
		return s.compareCounts(dstSchemaQualified, int64(numRows)) == nil
	})
	require.NoError(s.t, env.Error(s.t.Context()))

	var exists bool
	err = s.Conn().QueryRow(s.t.Context(),
		fmt.Sprintf("select exists(select * from %s where myreal <> 1729)", dstSchemaQualified)).Scan(&exists)
	require.NoError(s.t, err)
	require.False(s.t, exists)
}