feat(rest): add RegisterView to REST catalog (#753)

## Description:

Adds RegisterView(ctx, identifier, metadataLocation) to rest.Catalog,
the view equivalent of the existing RegisterTable. The method POSTs to
POST /v1/{prefix}/namespaces/{namespace}/register-view as defined in the
Iceberg REST catalog OpenAPI specification.

## Motivation

Fixes: #752

The only way to migrate a view between REST catalogs without this method
is LoadView + CreateView. That approach is lossy: CreateView resets
version-id to 1, reassigns default-catalog to the target catalog name,
and creates a new UUID — producing a new view rather than a registration
of the existing one. RegisterView avoids all of this by pointing the
target catalog at the existing metadata file, exactly as RegisterTable
does for tables.

## Changes

- catalog/rest/rest.go: RegisterView on *Catalog — POSTs {name,
metadata-location} to /namespaces/{ns}/register-view, parses the
response through the existing loadViewResponse path, maps 404 →
ErrNoSuchNamespace and 409 → ErrViewAlreadyExists.
- catalog/rest/rest_test.go: three test cases covering the 200, 404, and
409 responses, following the same structure as the existing
TestRegisterTable{200,404,409} tests.

## Related

- Java implementation: apache/iceberg#14868
- REST OpenAPI spec: open-api/rest-catalog-open-api.yaml — operationId
registerView
- Mailing list discussion:
https://lists.apache.org/thread/mxqz0fp6vz3nt7nl70wgrz6d2gr41rzj

Signed-off-by: Shubhendu Ram Tripathi <[email protected]>

Author: shtripat

catalog/rest/rest.go

@@ -1339,6 +1339,37 @@ type loadViewResponse struct {
 	Config      iceberg.Properties `json:"config"`
 }
 
+// RegisterView registers an existing view in the catalog using its metadata file location.
+// The metadata file must already be accessible to the catalog. This is the view equivalent
+// of RegisterTable, using the REST endpoint POST /namespaces/{ns}/register-view defined in
+// the Iceberg REST catalog specification.
+func (r *Catalog) RegisterView(ctx context.Context, identifier table.Identifier, metadataLoc string) (*view.View, error) {
+	ns, v, err := splitIdentForPath(identifier)
+	if err != nil {
+		return nil, err
+	}
+
+	type payload struct {
+		Name        string `json:"name"`
+		MetadataLoc string `json:"metadata-location"`
+	}
+
+	rsp, err := doPost[payload, loadViewResponse](ctx, r.baseURI, []string{"namespaces", ns, "register-view"},
+		payload{Name: v, MetadataLoc: metadataLoc}, r.cl, map[int]error{
+			http.StatusNotFound: catalog.ErrNoSuchNamespace, http.StatusConflict: catalog.ErrViewAlreadyExists,
+		})
+	if err != nil {
+		return nil, err
+	}
+
+	metadata, err := view.ParseMetadataBytes(rsp.RawMetadata)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse view metadata: %w", err)
+	}
+
+	return view.New(identifier, metadata, rsp.MetadataLoc), nil
+}
+
 // LoadView loads a view from the catalog.
 func (r *Catalog) LoadView(ctx context.Context, identifier table.Identifier) (*view.View, error) {
 	ns, v, err := splitIdentForPath(identifier)

catalog/rest/rest_test.go

@@ -2478,6 +2478,110 @@ func (r *RestCatalogSuite) TestCreateView404() {
 	r.ErrorIs(err, catalog.ErrNoSuchNamespace)
 }
 
+func (r *RestCatalogSuite) TestRegisterView200() {
+	const (
+		ns          = "fokko"
+		viewName    = "myview"
+		metadataLoc = "s3://bucket/warehouse/fokko.db/myview/metadata/00001.metadata.json"
+	)
+	identifier := table.Identifier{ns, viewName}
+
+	r.mux.HandleFunc("/v1/namespaces/"+ns+"/register-view", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodPost, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		var payload struct {
+			Name        string `json:"name"`
+			MetadataLoc string `json:"metadata-location"`
+		}
+		r.NoError(json.NewDecoder(req.Body).Decode(&payload))
+		r.Equal(viewName, payload.Name)
+		r.Equal(metadataLoc, payload.MetadataLoc)
+
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprintf(w, `{"metadata-location": %q, "metadata": %s, "config": {}}`,
+			metadataLoc, exampleViewMetadataJSON)
+	})
+
+	cat, err := rest.NewCatalog(context.Background(), "rest", r.srv.URL, rest.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	v, err := cat.RegisterView(context.Background(), identifier, metadataLoc)
+	r.Require().NoError(err)
+
+	r.Equal(identifier, v.Identifier())
+	r.Equal(metadataLoc, v.MetadataLocation())
+	r.Equal(uuid.MustParse("a1b2c3d4-e5f6-7890-1234-567890abcdef"), v.Metadata().ViewUUID())
+	r.EqualValues(1, v.Metadata().CurrentVersionID())
+	r.Equal(exampleViewSQL, v.Metadata().CurrentVersion().Representations[0].Sql)
+}
+
+func (r *RestCatalogSuite) TestRegisterView404() {
+	const (
+		ns          = "nonexistent"
+		viewName    = "myview"
+		metadataLoc = "s3://bucket/warehouse/nonexistent.db/myview/metadata/00001.metadata.json"
+	)
+
+	r.mux.HandleFunc("/v1/namespaces/"+ns+"/register-view", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodPost, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusNotFound)
+		json.NewEncoder(w).Encode(map[string]any{"error": errorResponse{
+			Message: "The given namespace does not exist",
+			Type:    "NoSuchNamespaceException",
+			Code:    404,
+		}})
+	})
+
+	cat, err := rest.NewCatalog(context.Background(), "rest", r.srv.URL, rest.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	_, err = cat.RegisterView(context.Background(), catalog.ToIdentifier(ns, viewName), metadataLoc)
+	r.ErrorIs(err, catalog.ErrNoSuchNamespace)
+	r.ErrorContains(err, "The given namespace does not exist")
+}
+
+func (r *RestCatalogSuite) TestRegisterView409() {
+	const (
+		ns          = "fokko"
+		viewName    = "alreadyexists"
+		metadataLoc = "s3://bucket/warehouse/fokko.db/alreadyexists/metadata/00001.metadata.json"
+	)
+
+	r.mux.HandleFunc("/v1/namespaces/"+ns+"/register-view", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodPost, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusConflict)
+		json.NewEncoder(w).Encode(map[string]any{"error": errorResponse{
+			Message: "The given view already exists",
+			Type:    "AlreadyExistsException",
+			Code:    409,
+		}})
+	})
+
+	cat, err := rest.NewCatalog(context.Background(), "rest", r.srv.URL, rest.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	_, err = cat.RegisterView(context.Background(), catalog.ToIdentifier(ns, viewName), metadataLoc)
+	r.ErrorIs(err, catalog.ErrViewAlreadyExists)
+	r.ErrorContains(err, "The given view already exists")
+}
+
 type mockTransport struct {
 	calls []struct {
 		method, path string