Method Test_Toast_Advance_3_SF

flow/e2e/snowflake_test.go:392–441

()

}

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

	srcTableName := s.attachSchemaSuffix("test_toast_sf_5")
	dstTableName := fmt.Sprintf("%s.%s", s.sfHelper.testSchemaName, "test_toast_sf_5")

	_, err := s.Conn().Exec(s.t.Context(), fmt.Sprintf(`
	CREATE TABLE IF NOT EXISTS %s (
			id SERIAL PRIMARY KEY,
			t1 text,
			t2 text,
			k int
	);
`, srcTableName))
	require.NoError(s.t, err)

	connectionGen := FlowConnectionGenerationConfig{
		FlowJobName:      s.attachSuffix("test_toast_sf_5"),
		TableNameMapping: map[string]string{srcTableName: dstTableName},
		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)
	/*
		transaction updating a single row
		multiple times with changed/unchanged toast columns
	*/
	_, err = s.Conn().Exec(s.t.Context(), fmt.Sprintf(`
		BEGIN;
		INSERT INTO %s (t1,t2,k) SELECT random_string(9000),random_string(9000),
		1 FROM generate_series(1,1);
		UPDATE %s SET k=102 WHERE id=1;
		UPDATE %s SET t1='dummy' WHERE id=1;
		UPDATE %s SET t2='dummy' WHERE id=1;
		END;
	`, srcTableName, srcTableName, srcTableName, srcTableName))
	EnvNoError(s.t, env, err)
	s.t.Log("Executed a transaction touching toast columns")

	EnvWaitForEqualTables(env, s, "normalizing tx", "test_toast_sf_5", `id,t1,t2,k`)
	env.Cancel(s.t.Context())

	RequireEnvCanceled(s.t, env)
}

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