feat(mc2mc): set execution project (#60)

feat: set execution project

mc2mc/internal/client/client.go

@@ -11,6 +11,7 @@ import (
 
 type OdpsClient interface {
 	ExecSQL(ctx context.Context, query string) error
+	SetDefaultProject(project string)
 }
 
 type Client struct {

mc2mc/internal/client/odps.go

@@ -54,6 +54,11 @@ func (c *odpsClient) ExecSQL(ctx context.Context, query string) error {
 	}
 }
 
+// SetDefaultProject sets the default project of the odps client
+func (c *odpsClient) SetDefaultProject(project string) {
+	c.client.SetDefaultProjectName(project)
+}
+
 // GetPartitionNames returns the partition names of the given table
 // by querying the table schema.
 func (c *odpsClient) GetPartitionNames(_ context.Context, tableID string) ([]string, error) {

mc2mc/internal/client/setup.go

@@ -10,6 +10,16 @@ import (
 
 type SetupFn func(c *Client) error
 
+func SetupDefaultProject(project string) SetupFn {
+	return func(c *Client) error {
+		if c.OdpsClient == nil {
+			return errors.New("odps client is required")
+		}
+		c.OdpsClient.SetDefaultProject(project)
+		return nil
+	}
+}
+
 func SetupLogger(logger *slog.Logger) SetupFn {
 	return func(c *Client) error {
 		c.logger = logger

mc2mc/internal/config/config.go

@@ -18,6 +18,7 @@ type ConfigEnv struct {
 	DestinationTableID        string `env:"DESTINATION_TABLE_ID"`
 	DStart                    string `env:"DSTART"`
 	DEnd                      string `env:"DEND"`
+	ExecutionProject          string `env:"EXECUTION_PROJECT"`
 	// TODO: delete this
 	DevEnablePartitionValue string `env:"DEV__ENABLE_PARTITION_VALUE" envDefault:"false"`
 	DevEnableAutoPartition  string `env:"DEV__ENABLE_AUTO_PARTITION" envDefault:"false"`

mc2mc/mc2mc.go

@@ -42,6 +42,7 @@ func mc2mc(envs []string) error {
 		client.SetupLogger(l),
 		client.SetupOTelSDK(cfg.OtelCollectorGRPCEndpoint, cfg.OtelAttributes),
 		client.SetupODPSClient(cfg.GenOdps()),
+		client.SetupDefaultProject(cfg.ExecutionProject),
 	)
 	if err != nil {
 		return errors.WithStack(err)