Method Test_Toast_SF

flow/e2e/snowflake_test.go:234–283

()

}

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

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

	_, 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_1"),
		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)
	/*
		Executing a transaction which
		1. changes both toast column
		2. changes no toast column
		2. changes 1 toast column
	*/
	_, 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,2);
		UPDATE %s SET k=102 WHERE id=1;
		UPDATE %s SET t1='dummy' WHERE id=2;
		END;
	`, 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_1", `id,t1,t2,k`)
	env.Cancel(s.t.Context())

	RequireEnvCanceled(s.t, env)
}

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