Method Test_Trips_Flow

flow/e2e/bigquery_source_test.go:412–483

()

}

func (s BigQueryClickhouseSuite) Test_Trips_Flow() {
	t := s.T()
	ctx := t.Context()

	source := s.Source().(*bigQuerySource)
	srcTable := "trips_1k"
	dstTable := "trips_1k_dst"

	t.Logf("ClickHouse database: %s", s.Peer().Config.(*protos.Peer_ClickhouseConfig).ClickhouseConfig.Database)

	count, err := source.helper.countRowsWithDataset(ctx, source.config.DatasetId, srcTable, "")
	require.NoError(t, err, "should be able to count rows in source table")
	require.Positive(t, count, "source table should have data")
	t.Logf("Source table %s has %d rows", srcTable, count)

	connectionGen := FlowConnectionGenerationConfig{
		FlowJobName: AddSuffix(s, srcTable),
		TableMappings: []*protos.TableMapping{
			{
				SourceTableIdentifier:      fmt.Sprintf("%s.%s", source.config.DatasetId, srcTable),
				DestinationTableIdentifier: s.DestinationTable(dstTable),
			},
		},
		Destination: s.Peer().Name,
	}
	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
	flowConnConfig.DoInitialSnapshot = true
	flowConnConfig.InitialSnapshotOnly = true
	flowConnConfig.SnapshotStagingPath = bigQueryTestStagingPath(s, srcTable)

	tc := NewTemporalClient(t)
	env := ExecutePeerflow(t, tc, flowConnConfig)

	EnvWaitForEqualTablesWithNames(
		env,
		s,
		"initial load to match",
		srcTable,
		dstTable,
		"trip_id,vendor_id,passenger_count,trip_distance,fare_amount",
	)

	EnvWaitForFinished(t, env, 3*time.Minute)

	apiClient, err := NewApiClient()
	require.NoError(t, err)

	statusResp, err := apiClient.MirrorStatus(ctx, &protos.MirrorStatusRequest{
		FlowJobName:     flowConnConfig.FlowJobName,
		IncludeFlowInfo: true,
	})
	require.NoError(t, err)

	cdcStatus := statusResp.GetCdcStatus()
	require.NotNil(t, cdcStatus)
	require.NotNil(t, cdcStatus.SnapshotStatus)
	require.NotEmpty(t, cdcStatus.SnapshotStatus.Clones)