This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 005d503a1 refactor(rest): share catalog object pagination requests
(#2284)
005d503a1 is described below
commit 005d503a1b8ac941531462249f0cb6a9eda19689
Author: Minh Vu <[email protected]>
AuthorDate: Fri Oct 23 22:12:46 2026 +0200
refactor(rest): share catalog object pagination requests (#2284)
* refactor(rest): share catalog object pagination requests
Signed-off-by: Minh Vu <[email protected]>
* test(rest): preserve 404 namespace error across listing kinds
* style(rest): gofmt identifier listing test
---------
Signed-off-by: Minh Vu <[email protected]>
Co-authored-by: Matt Topol <[email protected]>
---
catalog/rest/identifier_listing.go | 90 ++++++++++++
catalog/rest/identifier_listing_internal_test.go | 179 +++++++++++++++++++++++
catalog/rest/rest.go | 130 +---------------
3 files changed, 273 insertions(+), 126 deletions(-)
diff --git a/catalog/rest/identifier_listing.go
b/catalog/rest/identifier_listing.go
new file mode 100644
index 000000000..8dd59ad17
--- /dev/null
+++ b/catalog/rest/identifier_listing.go
@@ -0,0 +1,90 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package rest
+
+import (
+ "context"
+ "net/http"
+ "net/url"
+ "strconv"
+
+ "github.com/apache/iceberg-go/catalog"
+ "github.com/apache/iceberg-go/table"
+)
+
+type pagedIdentifierResponse[T any] struct {
+ Identifiers []T `json:"identifiers"`
+ NextPageToken string `json:"next-page-token,omitempty"`
+}
+
+// paginationQueryParams constructs the query shared by all paged REST
listings.
+// Non-positive page sizes and empty tokens are intentionally omitted.
+func paginationQueryParams(pageToken string, pageSize int) url.Values {
+ params := url.Values{}
+ if pageSize > 0 {
+ params.Set("pageSize", strconv.Itoa(pageSize))
+ }
+ if pageToken != "" {
+ params.Set("pageToken", pageToken)
+ }
+
+ return params
+}
+
+// listCatalogObjectsPage handles common REST listing behavior while keeping
the
+// table/view and function identifier wire shapes distinct through T.
+func listCatalogObjectsPage[T any](
+ r *Catalog, ctx context.Context, ep endpoint, namespace
table.Identifier,
+ pageToken string, pageSize int, convert func(T) table.Identifier,
+) ([]table.Identifier, string, error) {
+ if err := r.checkValidNamespace(namespace); err != nil {
+ return nil, "", err
+ }
+ // Missing advertised listing endpoints produce an empty result.
+ if !r.endpoints.allowed(ep) {
+ return nil, "", nil
+ }
+
+ path, err := ep.reqPath(r.encodeNamespace(namespace))
+ if err != nil {
+ return nil, "", err
+ }
+ uri := r.baseURI.JoinPath(path...)
+ uri.RawQuery = paginationQueryParams(pageToken, pageSize).Encode()
+
+ rsp, err := doGet[pagedIdentifierResponse[T]](ctx, uri, nil, r.cl,
+ map[int]error{http.StatusNotFound: catalog.ErrNoSuchNamespace})
+ if err != nil {
+ return nil, "", err
+ }
+
+ out := make([]table.Identifier, len(rsp.Identifiers))
+ for i, raw := range rsp.Identifiers {
+ out[i] = convert(raw)
+ }
+
+ return out, rsp.NextPageToken, nil
+}
+
+func objectIdentifier(id identifier) table.Identifier {
+ return append(id.Namespace, id.Name)
+}
+
+func functionIdentifier(levels []string) table.Identifier {
+ return table.Identifier(levels)
+}
diff --git a/catalog/rest/identifier_listing_internal_test.go
b/catalog/rest/identifier_listing_internal_test.go
new file mode 100644
index 000000000..ef5e30b60
--- /dev/null
+++ b/catalog/rest/identifier_listing_internal_test.go
@@ -0,0 +1,179 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package rest
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "reflect"
+ "sync/atomic"
+ "testing"
+
+ "github.com/apache/iceberg-go/catalog"
+ "github.com/apache/iceberg-go/table"
+)
+
+func TestCatalogObjectPagesPreserveWireShapesAndQueries(t *testing.T) {
+ for _, tc := range []struct {
+ name, route, response string
+ ep endpoint
+ want []table.Identifier
+ }{
+ {
+ name: "tables", route: "tables", ep: endpointListTables,
+ response:
`{"identifiers":[{"namespace":["analytics"],"name":"sales"},{"namespace":["analytics","raw"],"name":"events"}],"next-page-token":"next-page"}`,
+ want: []table.Identifier{{"analytics", "sales"},
{"analytics", "raw", "events"}},
+ },
+ {
+ name: "views", route: "views", ep: endpointListViews,
+ response:
`{"identifiers":[{"namespace":["analytics"],"name":"monthly"}],"next-page-token":"next-page"}`,
+ want: []table.Identifier{{"analytics", "monthly"}},
+ },
+ {
+ name: "functions", route: "functions", ep:
endpointListFunctions,
+ response:
`{"identifiers":[["analytics","normalize"],["analytics","helpers","parse"]],"next-page-token":"next-page"}`,
+ want: []table.Identifier{{"analytics",
"normalize"}, {"analytics", "helpers", "parse"}},
+ },
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ var requests atomic.Int32
+ server := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, req *http.Request) {
+ requests.Add(1)
+ if req.Method != http.MethodGet {
+ t.Errorf("method = %q, want GET",
req.Method)
+ }
+ if wantPath := "/v1/namespaces/analytics/" +
tc.route; req.URL.Path != wantPath {
+ t.Errorf("path = %q, want %q",
req.URL.Path, wantPath)
+ }
+ if got := req.URL.Query().Get("pageToken"); got
!= "token +/%" {
+ t.Errorf("pageToken = %q, want %q",
got, "token +/%")
+ }
+ if got := req.URL.Query().Get("pageSize"); got
!= "37" {
+ t.Errorf("pageSize = %q, want 37", got)
+ }
+ _, _ = w.Write([]byte(tc.response))
+ }))
+ defer server.Close()
+
+ baseURI, err := url.Parse(server.URL + "/v1")
+ if err != nil {
+ t.Fatal(err)
+ }
+ cat := &Catalog{baseURI: baseURI, cl: server.Client(),
endpoints: newEndpointSet([]endpoint{tc.ep})}
+ namespace := table.Identifier{"analytics"}
+ var got []table.Identifier
+ var nextToken string
+ switch tc.ep {
+ case endpointListTables:
+ got, nextToken, err =
cat.listTablesPage(context.Background(), namespace, "token +/%", 37)
+ case endpointListViews:
+ got, nextToken, err =
cat.listViewsPage(context.Background(), namespace, "token +/%", 37)
+ case endpointListFunctions:
+ got, nextToken, err =
cat.listFunctionsPage(context.Background(), namespace, "token +/%", 37)
+ default:
+ t.Fatalf("unexpected endpoint: %v", tc.ep)
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !reflect.DeepEqual(got, tc.want) {
+ t.Errorf("identifiers = %v, want %v", got,
tc.want)
+ }
+ if nextToken != "next-page" {
+ t.Errorf("next token = %q, want next-page",
nextToken)
+ }
+ if requests.Load() != 1 {
+ t.Errorf("requests = %d, want 1",
requests.Load())
+ }
+ })
+ }
+}
+
+func TestCatalogObjectPagesDoNotFetchUnsupportedEndpoints(t *testing.T) {
+ var requests atomic.Int32
+ server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter,
*http.Request) { requests.Add(1) }))
+ defer server.Close()
+ baseURI, err := url.Parse(server.URL + "/v1")
+ if err != nil {
+ t.Fatal(err)
+ }
+ cat := &Catalog{baseURI: baseURI, cl: server.Client(), endpoints:
newEndpointSet(nil)}
+
+ for _, listPage := range []func(context.Context, table.Identifier,
string, int) ([]table.Identifier, string, error){
+ cat.listTablesPage, cat.listViewsPage, cat.listFunctionsPage,
+ } {
+ identifiers, token, err := listPage(context.Background(),
table.Identifier{"analytics"}, "", 0)
+ if err != nil || len(identifiers) != 0 || token != "" {
+ t.Errorf("unsupported listing returned (%v, %q, %v),
want empty result", identifiers, token, err)
+ }
+ _, _, err = listPage(context.Background(), nil, "", 0)
+ if err == nil {
+ t.Error("invalid namespace must be rejected before
endpoint gating")
+ }
+ }
+ if requests.Load() != 0 {
+ t.Errorf("requests = %d, want 0", requests.Load())
+ }
+}
+
+func TestCatalogObjectPagesMapNotFoundToMissingNamespace(t *testing.T) {
+ server := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, _ *http.Request) {
+ w.WriteHeader(http.StatusNotFound)
+ }))
+ defer server.Close()
+
+ baseURI, err := url.Parse(server.URL + "/v1")
+ if err != nil {
+ t.Fatal(err)
+ }
+ cat := &Catalog{
+ baseURI: baseURI,
+ cl: server.Client(),
+ endpoints: newEndpointSet([]endpoint{
+ endpointListTables, endpointListViews,
endpointListFunctions,
+ }),
+ }
+ for _, listPage := range []func(context.Context, table.Identifier,
string, int) ([]table.Identifier, string, error){
+ cat.listTablesPage, cat.listViewsPage, cat.listFunctionsPage,
+ } {
+ _, _, err := listPage(t.Context(), table.Identifier{"missing"},
"", 0)
+ if !errors.Is(err, catalog.ErrNoSuchNamespace) {
+ t.Errorf("404 should indicate a missing namespace, got
%v", err)
+ }
+ }
+}
+
+func TestPaginationQueryParams(t *testing.T) {
+ for _, tc := range []struct {
+ token string
+ size int
+ want string
+ }{
+ {"", 0, ""},
+ {"", -1, ""},
+ {"abc", 0, "pageToken=abc"},
+ {"a+b /%", 42, "pageSize=42&pageToken=a%2Bb+%2F%25"},
+ } {
+ if got := paginationQueryParams(tc.token, tc.size).Encode();
got != tc.want {
+ t.Errorf("paginationQueryParams(%q, %d) = %q, want %q",
tc.token, tc.size, got, tc.want)
+ }
+ }
+}
diff --git a/catalog/rest/rest.go b/catalog/rest/rest.go
index fcd5d8aa7..a2f175206 100644
--- a/catalog/rest/rest.go
+++ b/catalog/rest/rest.go
@@ -31,7 +31,6 @@ import (
"net/http"
"net/url"
"slices"
- "strconv"
"strings"
"sync"
"sync/atomic"
@@ -1536,43 +1535,7 @@ func (r *Catalog) ListTables(ctx context.Context,
namespace table.Identifier) it
}
func (r *Catalog) listTablesPage(ctx context.Context, namespace
table.Identifier, pageToken string, pageSize int) ([]table.Identifier, string,
error) {
- if err := r.checkValidNamespace(namespace); err != nil {
- return nil, "", err
- }
- // Unsupported listing yields an empty result rather than an error.
- if !r.endpoints.allowed(endpointListTables) {
- return nil, "", nil
- }
- ns := r.encodeNamespace(namespace)
- path, err := endpointListTables.reqPath(ns)
- if err != nil {
- return nil, "", err
- }
- uri := r.baseURI.JoinPath(path...)
-
- v := url.Values{}
- if pageSize > 0 {
- v.Set("pageSize", strconv.Itoa(pageSize))
- }
- if pageToken != "" {
- v.Set("pageToken", pageToken)
- }
-
- uri.RawQuery = v.Encode()
- type resp struct {
- Identifiers []identifier `json:"identifiers"`
- NextPageToken string `json:"next-page-token,omitempty"`
- }
- rsp, err := doGet[resp](ctx, uri, []string{}, r.cl,
map[int]error{http.StatusNotFound: catalog.ErrNoSuchNamespace})
- if err != nil {
- return nil, "", err
- }
- out := make([]table.Identifier, len(rsp.Identifiers))
- for i, id := range rsp.Identifiers {
- out[i] = append(id.Namespace, id.Name)
- }
-
- return out, rsp.NextPageToken, nil
+ return listCatalogObjectsPage(r, ctx, endpointListTables, namespace,
pageToken, pageSize, objectIdentifier)
}
func (r *Catalog) nsSeparator() string {
@@ -2291,16 +2254,10 @@ func (r *Catalog) listNamespacesPage(ctx
context.Context, parent table.Identifie
uri := r.baseURI.JoinPath(path...)
- v := url.Values{}
+ v := paginationQueryParams(pageToken, pageSize)
if len(parent) != 0 {
v.Set("parent", r.namespaceToQueryParam(parent))
}
- if pageSize > 0 {
- v.Set("pageSize", strconv.Itoa(pageSize))
- }
- if pageToken != "" {
- v.Set("pageToken", pageToken)
- }
if len(v) > 0 {
uri.RawQuery = v.Encode()
}
@@ -2469,45 +2426,7 @@ func (r *Catalog) ListViews(ctx context.Context,
namespace table.Identifier) ite
}
func (r *Catalog) listViewsPage(ctx context.Context, namespace
table.Identifier, pageToken string, pageSize int) ([]table.Identifier, string,
error) {
- if err := r.checkValidNamespace(namespace); err != nil {
- return nil, "", err
- }
- // Unsupported listing yields an empty result rather than an error.
- if !r.endpoints.allowed(endpointListViews) {
- return nil, "", nil
- }
- ns := r.encodeNamespace(namespace)
- path, err := endpointListViews.reqPath(ns)
- if err != nil {
- return nil, "", err
- }
- uri := r.baseURI.JoinPath(path...)
-
- v := url.Values{}
- if pageSize > 0 {
- v.Set("pageSize", strconv.Itoa(pageSize))
- }
- if pageToken != "" {
- v.Set("pageToken", pageToken)
- }
-
- uri.RawQuery = v.Encode()
- type resp struct {
- Identifiers []identifier `json:"identifiers"`
- NextPageToken string `json:"next-page-token,omitempty"`
- }
-
- rsp, err := doGet[resp](ctx, uri, []string{}, r.cl,
map[int]error{http.StatusNotFound: catalog.ErrNoSuchNamespace})
- if err != nil {
- return nil, "", err
- }
-
- out := make([]table.Identifier, len(rsp.Identifiers))
- for i, id := range rsp.Identifiers {
- out[i] = append(id.Namespace, id.Name)
- }
-
- return out, rsp.NextPageToken, nil
+ return listCatalogObjectsPage(r, ctx, endpointListViews, namespace,
pageToken, pageSize, objectIdentifier)
}
func (r *Catalog) getPageSize(ctx context.Context) int {
@@ -2875,48 +2794,7 @@ func (r *Catalog) ListFunctions(ctx context.Context,
namespace table.Identifier)
}
func (r *Catalog) listFunctionsPage(ctx context.Context, namespace
table.Identifier, pageToken string, pageSize int) ([]table.Identifier, string,
error) {
- if err := r.checkValidNamespace(namespace); err != nil {
- return nil, "", err
- }
- // Unsupported listing yields an empty result rather than an error.
- if !r.endpoints.allowed(endpointListFunctions) {
- return nil, "", nil
- }
- ns := r.encodeNamespace(namespace)
- path, err := endpointListFunctions.reqPath(ns)
- if err != nil {
- return nil, "", err
- }
- uri := r.baseURI.JoinPath(path...)
-
- v := url.Values{}
- if pageSize > 0 {
- v.Set("pageSize", strconv.Itoa(pageSize))
- }
- if pageToken != "" {
- v.Set("pageToken", pageToken)
- }
-
- uri.RawQuery = v.Encode()
- // Function identifiers are CatalogObjectIdentifier values: ordered
- // hierarchy levels (the namespace levels followed by the function
- // name), not the {namespace, name} object tables and views use.
- type resp struct {
- Identifiers [][]string `json:"identifiers"`
- NextPageToken string `json:"next-page-token,omitempty"`
- }
-
- rsp, err := doGet[resp](ctx, uri, []string{}, r.cl,
map[int]error{http.StatusNotFound: catalog.ErrNoSuchNamespace})
- if err != nil {
- return nil, "", err
- }
-
- out := make([]table.Identifier, len(rsp.Identifiers))
- for i, levels := range rsp.Identifiers {
- out[i] = table.Identifier(levels)
- }
-
- return out, rsp.NextPageToken, nil
+ return listCatalogObjectsPage(r, ctx, endpointListFunctions, namespace,
pageToken, pageSize, functionIdentifier)
}
// loadFunctionResponse contains the response from loading a function.
