Change ALTER RELATION semantics (#1557)

Author: EinKrebs

coordinator/mock/mock_coordinator.go

@@ -94,6 +94,18 @@ func (d *DistributionsServer) AlterDistributedRelation(ctx context.Context, req
 	return nil, d.impl.AlterDistributedRelation(ctx, req.GetId(), ds)
 }
 
+func (d *DistributionsServer) AlterDistributedRelationSchema(ctx context.Context, req *protos.AlterDistributedRelationSchemaRequest) (*emptypb.Empty, error) {
+	return nil, d.impl.AlterDistributedRelationSchema(ctx, req.GetId(), req.GetRelationName(), req.GetSchemaName())
+}
+
+func (d *DistributionsServer) AlterDistributedRelationDistributionKey(ctx context.Context, req *protos.AlterDistributedRelationDistributionKeyRequest) (*emptypb.Empty, error) {
+	key, err := distributions.DistributionKeyFromProto(req.GetDistributionKey())
+	if err != nil {
+		return nil, err
+	}
+	return nil, d.impl.AlterDistributedRelationDistributionKey(ctx, req.GetId(), req.GetRelationName(), key)
+}
+
 func (d *DistributionsServer) GetDistribution(ctx context.Context, req *protos.GetDistributionRequest) (*protos.GetDistributionReply, error) {
 	ds, err := d.impl.GetDistribution(ctx, req.GetId())
 	if err != nil {

coordinator/provider/distributions.go

@@ -849,6 +849,46 @@ func (a *Adapter) AlterDistributedRelation(ctx context.Context, id string, rel *
 	return err
 }
 
+// AlterDistributedRelationSchema alters the sequence name of a distributed relation.
+//
+// Parameters:
+// - ctx (context.Context): The context for the request.
+// - id (string): The ID of the distribution of the relation.
+// - relName (string): The name of the relation.
+// - schemaName (string): the new schema name for the relation.
+//
+// Returns:
+// - error: An error if the alteration of the distribution's attachments fails, otherwise nil.
+func (a *Adapter) AlterDistributedRelationSchema(ctx context.Context, id string, relName string, schemaName string) error {
+	c := proto.NewDistributionServiceClient(a.conn)
+	_, err := c.AlterDistributedRelationSchema(ctx, &proto.AlterDistributedRelationSchemaRequest{
+		Id:           id,
+		RelationName: relName,
+		SchemaName:   schemaName,
+	})
+	return err
+}
+
+// AlterDistributedRelationDistributionKey alters the distribution key metadata of a distributed relation.
+//
+// Parameters:
+// - ctx (context.Context): The context for the request.
+// - id (string): The ID of the distribution of the relation.
+// - relName (string): The name of the relation.
+// - distributionKey ([]distributions.DistributionKeyEntry): the new distribution key for the relation.
+//
+// Returns:
+// - error: An error if the alteration of the distribution's attachments fails, otherwise nil.
+func (a *Adapter) AlterDistributedRelationDistributionKey(ctx context.Context, id string, relName string, distributionKey []distributions.DistributionKeyEntry) error {
+	c := proto.NewDistributionServiceClient(a.conn)
+	_, err := c.AlterDistributedRelationDistributionKey(ctx, &proto.AlterDistributedRelationDistributionKeyRequest{
+		Id:              id,
+		RelationName:    relName,
+		DistributionKey: distributions.DistributionKeyToProto(distributionKey),
+	})
+	return err
+}
+
 // AlterDistributionDetach detaches a relation from a distribution using the provided ID and relation name.
 //
 // Parameters:

pkg/coord/adapter.go

@@ -2101,6 +2101,56 @@ func (qc *ClusteredCoordinator) AlterDistributedRelation(ctx context.Context, id
 	})
 }
 
+// AlterDistributedRelationSchema changes the schema name of a relation attached to a distribution
+// TODO: unit tests
+func (qc *ClusteredCoordinator) AlterDistributedRelationSchema(ctx context.Context, id string, relName string, schemaName string) error {
+	if err := qc.Coordinator.AlterDistributedRelationSchema(ctx, id, relName, schemaName); err != nil {
+		return err
+	}
+
+	return qc.traverseRouters(ctx, func(cc *grpc.ClientConn) error {
+		cl := proto.NewDistributionServiceClient(cc)
+		resp, err := cl.AlterDistributedRelationSchema(context.TODO(), &proto.AlterDistributedRelationSchemaRequest{
+			Id:           id,
+			RelationName: relName,
+			SchemaName:   schemaName,
+		})
+		if err != nil {
+			return err
+		}
+
+		spqrlog.Zero.Debug().
+			Interface("response", resp).
+			Msg("alter relation schema response")
+		return nil
+	})
+}
+
+// AlterDistributedRelationSchema changes the distribution key of a relation attached to a distribution
+// TODO: unit tests
+func (qc *ClusteredCoordinator) AlterDistributedRelationDistributionKey(ctx context.Context, id string, relName string, distributionKey []distributions.DistributionKeyEntry) error {
+	if err := qc.Coordinator.AlterDistributedRelationDistributionKey(ctx, id, relName, distributionKey); err != nil {
+		return err
+	}
+
+	return qc.traverseRouters(ctx, func(cc *grpc.ClientConn) error {
+		cl := proto.NewDistributionServiceClient(cc)
+		resp, err := cl.AlterDistributedRelationDistributionKey(context.TODO(), &proto.AlterDistributedRelationDistributionKeyRequest{
+			Id:              id,
+			RelationName:    relName,
+			DistributionKey: distributions.DistributionKeyToProto(distributionKey),
+		})
+		if err != nil {
+			return err
+		}
+
+		spqrlog.Zero.Debug().
+			Interface("response", resp).
+			Msg("alter relation distribution key response")
+		return nil
+	})
+}
+
 func (qc *ClusteredCoordinator) DropSequence(ctx context.Context, seqName string, force bool) error {
 	if err := qc.Coordinator.DropSequence(ctx, seqName, force); err != nil {
 		return err

pkg/coord/clustered_coord.go

@@ -95,6 +95,26 @@ func (lc *Coordinator) AlterDistributedRelation(ctx context.Context, id string,
 	return nil
 }
 
+// AlterDistributedRelationDistributionKey implements meta.EntityMgr.
+func (lc *Coordinator) AlterDistributedRelationDistributionKey(ctx context.Context, id string, relName string, distributionKey []distributions.DistributionKeyEntry) error {
+	if id == distributions.REPLICATED {
+		return fmt.Errorf("setting distribution key is forbidden for reference relations")
+	}
+	ds, err := lc.GetDistribution(ctx, id)
+	if err != nil {
+		return err
+	}
+	if len(ds.ColTypes) != len(distributionKey) {
+		return fmt.Errorf("cannot alter relation \"%s\" distribution key: numbers of columns mismatch", relName)
+	}
+	return lc.qdb.AlterDistributedRelationDistributionKey(ctx, id, relName, distributions.DistributionKeyToDB(distributionKey))
+}
+
+// AlterDistributedRelationSchema implements meta.EntityMgr.
+func (lc *Coordinator) AlterDistributedRelationSchema(ctx context.Context, id string, relName string, schemaName string) error {
+	return lc.qdb.AlterDistributedRelationSchema(ctx, id, relName, schemaName)
+}
+
 // BatchMoveKeyRange implements meta.EntityMgr.
 func (lc *Coordinator) BatchMoveKeyRange(ctx context.Context, req *kr.BatchMoveKeyRange) error {
 	panic("unimplemented")

pkg/coord/coord.go

@@ -400,7 +400,10 @@ func processCreate(ctx context.Context, astmt spqrparser.Statement, mngr EntityM
 func processAlter(ctx context.Context, astmt spqrparser.Statement, mngr EntityMgr, cli *clientinteractor.PSQLInteractor) error {
 	switch stmt := astmt.(type) {
 	case *spqrparser.AlterDistribution:
-		return processAlterDistribution(ctx, stmt.Element, mngr, cli)
+		if stmt.Distribution == nil {
+			return fmt.Errorf("failed to process 'ALTER DISTRIBUTION' statement: distribution ID is nil")
+		}
+		return processAlterDistribution(ctx, stmt.Element, mngr, cli, stmt.Distribution.ID)
 	default:
 		return ErrUnknownCoordinatorCommand
 	}
@@ -416,7 +419,7 @@ func processAlter(ctx context.Context, astmt spqrparser.Statement, mngr EntityMg
 //
 // Returns:
 // - error: An error if the operation fails, otherwise nil.
-func processAlterDistribution(ctx context.Context, astmt spqrparser.Statement, mngr EntityMgr, cli *clientinteractor.PSQLInteractor) error {
+func processAlterDistribution(ctx context.Context, astmt spqrparser.Statement, mngr EntityMgr, cli *clientinteractor.PSQLInteractor, dsId string) error {
 	switch stmt := astmt.(type) {
 	case *spqrparser.AttachRelation:
 
@@ -426,7 +429,7 @@ func processAlterDistribution(ctx context.Context, astmt spqrparser.Statement, m
 			rels = append(rels, distributions.DistributedRelationFromSQL(drel))
 		}
 
-		if stmt.Distribution.ID == "default" {
+		if dsId == "default" {
 			list, err := mngr.ListDistributions(ctx)
 			if err != nil {
 				return spqrerror.New(spqrerror.SPQR_OBJECT_NOT_EXIST, "error while selecting list of distributions")
@@ -437,29 +440,29 @@ func processAlterDistribution(ctx context.Context, astmt spqrparser.Statement, m
 			if len(list) > 1 {
 				return spqrerror.New(spqrerror.SPQR_OBJECT_NOT_EXIST, "distributions count not equal one, use FOR DISTRIBUTION syntax")
 			}
-			stmt.Distribution.ID = list[0].Id
+			dsId = list[0].Id
 		}
 
-		selectedDistribId := stmt.Distribution.ID
+		selectedDistribId := dsId
 
 		if err := mngr.AlterDistributionAttach(ctx, selectedDistribId, rels); err != nil {
 			return cli.ReportError(err)
 		}
 
 		return cli.AlterDistributionAttach(ctx, selectedDistribId, rels)
 	case *spqrparser.DetachRelation:
-		if err := mngr.AlterDistributionDetach(ctx, stmt.Distribution.ID, stmt.RelationName); err != nil {
+		if err := mngr.AlterDistributionDetach(ctx, dsId, stmt.RelationName); err != nil {
 			return err
 		}
-		return cli.AlterDistributionDetach(ctx, stmt.Distribution.ID, stmt.RelationName.String())
+		return cli.AlterDistributionDetach(ctx, dsId, stmt.RelationName.String())
 	case *spqrparser.AlterRelation:
-		if err := mngr.AlterDistributedRelation(ctx, stmt.Distribution.ID, distributions.DistributedRelationFromSQL(stmt.Relation)); err != nil {
+		if err := mngr.AlterDistributedRelation(ctx, dsId, distributions.DistributedRelationFromSQL(stmt.Relation)); err != nil {
 			return err
 		}
 		qName := rfqn.RelationFQN{RelationName: stmt.Relation.Name, SchemaName: stmt.Relation.SchemaName}
-		return cli.AlterDistributedRelation(ctx, stmt.Distribution.ID, qName.String())
+		return cli.AlterDistributedRelation(ctx, dsId, qName.String())
 	case *spqrparser.DropDefaultShard:
-		if distribution, err := mngr.GetDistribution(ctx, stmt.Distribution.ID); err != nil {
+		if distribution, err := mngr.GetDistribution(ctx, dsId); err != nil {
 			return err
 		} else {
 			manager := NewDefaultShardManager(distribution, mngr)
@@ -470,7 +473,7 @@ func processAlterDistribution(ctx context.Context, astmt spqrparser.Statement, m
 			}
 		}
 	case *spqrparser.AlterDefaultShard:
-		if distribution, err := mngr.GetDistribution(ctx, stmt.Distribution.ID); err != nil {
+		if distribution, err := mngr.GetDistribution(ctx, dsId); err != nil {
 			return err
 		} else {
 			manager := NewDefaultShardManager(distribution, mngr)
@@ -479,11 +482,43 @@ func processAlterDistribution(ctx context.Context, astmt spqrparser.Statement, m
 			}
 			return cli.MakeSimpleResponse(ctx, manager.SuccessCreateResponse(stmt.Shard))
 		}
+	case *spqrparser.AlterRelationV2:
+		return processAlterRelation(ctx, stmt.Element, mngr, cli, dsId, stmt.RelationName)
 	default:
 		return ErrUnknownCoordinatorCommand
 	}
 }
 
+// processAlterDistribution processes the given 'ALTER DISTRIBUTION ALTER RELATION' statement and performs the corresponding operation.
+//
+// Parameters:
+// - ctx (context.Context): The context for the operation.
+// - astmt (spqrparser.Statement): The alter relation statement to be processed.
+// - mngr (EntityMgr): The entity manager for performing the operation.
+// - cli (*clientinteractor.PSQLInteractor): The PSQL client interactor for interacting with the PSQL server.
+// - dsId (string): ID of the distribution, to which the relation belongs.
+// - relName (string): the name of the relation to alter.
+//
+// Returns:
+// - error: An error if the operation fails, otherwise nil.
+func processAlterRelation(ctx context.Context, astmt spqrparser.Statement, mngr EntityMgr, cli *clientinteractor.PSQLInteractor, dsId string, relName string) error {
+	switch stmt := astmt.(type) {
+	case *spqrparser.AlterRelationSchema:
+		if err := mngr.AlterDistributedRelationSchema(ctx, dsId, relName, stmt.SchemaName); err != nil {
+			return err
+		}
+		qName := rfqn.RelationFQN{RelationName: relName, SchemaName: stmt.SchemaName}
+		return cli.AlterDistributedRelation(ctx, dsId, qName.String())
+	case *spqrparser.AlterRelationDistributionKey:
+		if err := mngr.AlterDistributedRelationDistributionKey(ctx, dsId, relName, distributions.DistributionKeyFromSQL(stmt.DistributionKey)); err != nil {
+			return err
+		}
+		return cli.AlterDistributedRelation(ctx, dsId, relName)
+	default:
+		return fmt.Errorf("unexpected 'ALTER RELATION' request type %T", stmt)
+	}
+}
+
 // TODO : unit tests
 
 // ProcMetadataCommand processes various coordinator commands based on the provided statement.