Method Test_Multi_Table_SF

flow/e2e/snowflake_test.go:550–600

()

}

func (s PeerFlowE2ETestSuiteSF) Test_Multi_Table_SF() {
	tc := NewTemporalClient(s.t)

	srcTable1Name := s.attachSchemaSuffix("test1_sf")
	srcTable2Name := s.attachSchemaSuffix("test2_sf")
	dstTable1Name := fmt.Sprintf("%s.%s", s.sfHelper.testSchemaName, "test1_sf")
	dstTable2Name := fmt.Sprintf("%s.%s", s.sfHelper.testSchemaName, "test2_sf")

	_, err := s.Conn().Exec(s.t.Context(), fmt.Sprintf(`
	CREATE TABLE IF NOT EXISTS %s (id serial primary key, c1 int, c2 text);
	CREATE TABLE IF NOT EXISTS %s (id serial primary key, c1 int, c2 text);
	`, srcTable1Name, srcTable2Name))
	require.NoError(s.t, err)

	connectionGen := FlowConnectionGenerationConfig{
		FlowJobName:      s.attachSuffix("test_multi_table"),
		TableNameMapping: map[string]string{srcTable1Name: dstTable1Name, srcTable2Name: dstTable2Name},
		Destination:      s.Peer().Name,
	}

	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
	flowConnConfig.MaxBatchSize = 100

	// wait for PeerFlowStatusQuery to finish setup
	// and execute a transaction touching toast columns
	env := ExecutePeerflow(s.t, tc, flowConnConfig)
	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
	/* inserting across multiple tables*/
	_, err = s.Conn().Exec(s.t.Context(), fmt.Sprintf(`
	INSERT INTO %s (c1,c2) VALUES (1,'dummy_1');
	INSERT INTO %s (c1,c2) VALUES (-1,'dummy_-1');
	`, srcTable1Name, srcTable2Name))
	EnvNoError(s.t, env, err)

	EnvWaitFor(s.t, env, 2*time.Minute, "normalize both tables", func() bool {
		count1, err := s.sfHelper.CountRows(s.t.Context(), "test1_sf")
		if err != nil {
			return false
		}
		count2, err := s.sfHelper.CountRows(s.t.Context(), "test2_sf")
		if err != nil {
			return false
		}

		return count1 == 1 && count2 == 1
	})

	env.Cancel(s.t.Context())

	RequireEnvCanceled(s.t, env)
}

func (s PeerFlowE2ETestSuiteSF) Test_Composite_PKey_SF() {
	tc := NewTemporalClient(s.t)