Method PartialWrite

internal/servers/schema_server.go:92–200

PartialWrite applies incremental updates to the schema of a specific tenant based on the provided request.

(ctx context.Context, request *v1.SchemaPartialWriteRequest)

// PartialWrite applies incremental updates to the schema of a specific tenant based on the provided request.
func (r *SchemaServer) PartialWrite(ctx context.Context, request *v1.SchemaPartialWriteRequest) (*v1.SchemaPartialWriteResponse, error) {
	// Start a new tracing span for monitoring and observability.
	ctx, span := internal.Tracer.Start(ctx, "schemas.partial-write")
	defer span.End() // Ensure the span is closed at the end of the function.

	// Retrieve or default the schema version from the request.
	version := request.GetMetadata().GetSchemaVersion()
	if version == "" { // If not provided, fetch the latest version.
		ver, err := r.sr.HeadVersion(ctx, request.GetTenantId())
		if err != nil {
			return nil, status.Error(GetStatus(err), err.Error()) // Return version error
		}
		version = ver
	}

	// Fetch the current schema definition as a string.
	definitions, err := r.sr.ReadSchemaString(ctx, request.GetTenantId(), version)
	if err != nil {
		span.RecordError(err) // Log and record the error.
		return nil, status.Error(GetStatus(err), err.Error())
	}

	// Parse the schema definitions into a structured format.
	p := parser.NewParser(strings.Join(definitions, "\n"))
	schema, err := p.Parse()
	if err != nil {
		span.RecordError(err) // Log and record the error.
		return nil, status.Error(GetStatus(err), err.Error())
	}

	// Iterate through each partial update in the request and apply changes.
	for entityName, partials := range request.GetPartials() {
		for _, write := range partials.GetWrite() { // Handle new schema statements.
			pr := parser.NewParser(write)
			stmt, err := pr.ParsePartial(entityName)
			if err != nil {
				span.RecordError(err)
				return nil, status.Error(GetStatus(err), err.Error())
			}
			err = schema.AddStatement(entityName, stmt)
			if err != nil {
				span.RecordError(err)
				return nil, status.Error(GetStatus(err), err.Error())
			}
		}

		for _, update := range partials.GetUpdate() { // Handle schema updates.
			pr := parser.NewParser(update)
			stmt, err := pr.ParsePartial(entityName)
			if err != nil {
				span.RecordError(err)
				return nil, status.Error(GetStatus(err), err.Error())
			}
			err = schema.UpdateStatement(entityName, stmt)
			if err != nil {
				span.RecordError(err)
				return nil, status.Error(GetStatus(err), err.Error())
			}