Stabilize NetSuite SuiteQL pagination with best-effort ORDER BY

Offset/limit pagination over SuiteQL is only deterministic with a stable
ORDER BY. Plain table reads now sort by the incremental key (when set)
followed by the primary keys as a unique tie-breaker, so multi-page
extracts no longer skip or duplicate rows.

Ordering is best-effort and never blocks a load: if NetSuite rejects the
auto-added ORDER BY with a 400 (e.g. a table that exposes no sortable
column), the read retries once without it, warns that multi-page results
may be inconsistent, and continues. Custom (query:) reads are sent as-is.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: karakanb

docs/supported-sources/netsuite.md

@@ -65,6 +65,6 @@ ingestr ingest \
   --dest-table "main.netsuite_transactions"
 ```
 
-For plain table names, ingestr runs `SELECT * FROM <source-table>` through SuiteQL. For joins, selected columns, built-in functions, or aliases, use the `query:` source-table form.
+For plain table names, ingestr runs `SELECT * FROM <source-table>` through SuiteQL, appending an `ORDER BY` on the primary key (and incremental key when set) so that offset-based pagination returns a stable result set. If the table exposes no such column, ingestr retries without the `ORDER BY` and still loads the data, warning that results spanning more than one page (1000 rows) may be inconsistent; pass `--primary-key` to point at a sortable unique key. For joins, selected columns, built-in functions, or aliases, use the `query:` source-table form; custom queries are sent as-is, so add your own `ORDER BY` when the result spans more than one page.
 
 NetSuite SuiteQL responses do not expose a static schema through this connector, so ingestr infers the schema from extracted rows.

pkg/source/netsuite/client.go

@@ -266,6 +266,18 @@ type SuiteQLResponse struct {
 	Items        []map[string]interface{} `json:"items"`
 }
 
+// SuiteQLError is returned when the SuiteQL endpoint responds with a non-success
+// status. StatusCode lets callers distinguish client errors (e.g. a 400 from an
+// invalid ORDER BY column) from transient failures.
+type SuiteQLError struct {
+	StatusCode int
+	Body       string
+}
+
+func (e *SuiteQLError) Error() string {
+	return fmt.Sprintf("netsuite suiteql returned status %d: %s", e.StatusCode, e.Body)
+}
+
 func (c *Client) SuiteQL(ctx context.Context, query string, limit, offset int) (*SuiteQLResponse, error) {
 	if limit <= 0 {
 		limit = maxSuiteQLPageSize
@@ -294,7 +306,7 @@ func (c *Client) SuiteQL(ctx context.Context, query string, limit, offset int) (
 		return nil, fmt.Errorf("netsuite suiteql request failed: %w", err)
 	}
 	if !resp.IsSuccess() {
-		return nil, fmt.Errorf("netsuite suiteql returned status %d: %s", resp.StatusCode(), resp.String())
+		return nil, &SuiteQLError{StatusCode: resp.StatusCode(), Body: resp.String()}
 	}
 
 	var out SuiteQLResponse

pkg/source/netsuite/netsuite.go

@@ -2,7 +2,9 @@ package netsuite
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"net/http"
 	"net/url"
 	"strings"
 	"time"
@@ -93,14 +95,14 @@ func (s *NetSuiteSource) GetTable(ctx context.Context, req source.TableRequest)
 			return nil, fmt.Errorf("netsuite source does not have a predefined schema; schema inference is required")
 		},
 		ReadFn: func(ctx context.Context, opts source.ReadOptions) (<-chan source.RecordBatchResult, error) {
-			return s.readTable(ctx, tableName, opts)
+			return s.readTable(ctx, tableName, primaryKeys, opts)
 		},
 	}, nil
 }
 
-func (s *NetSuiteSource) readTable(ctx context.Context, tableName string, opts source.ReadOptions) (<-chan source.RecordBatchResult, error) {
-	query := buildSuiteQL(tableName, opts)
-	return s.readSuiteQL(ctx, query, opts)
+func (s *NetSuiteSource) readTable(ctx context.Context, tableName string, primaryKeys []string, opts source.ReadOptions) (<-chan source.RecordBatchResult, error) {
+	query, fallback := buildSuiteQL(tableName, primaryKeys, opts)
+	return s.readSuiteQL(ctx, query, fallback, opts)
 }
 
 func (s *NetSuiteSource) ExecuteCustomQuery(ctx context.Context, query string, opts source.ReadOptions) (<-chan source.RecordBatchResult, error) {
@@ -109,10 +111,10 @@ func (s *NetSuiteSource) ExecuteCustomQuery(ctx context.Context, query string, o
 	if query == "" {
 		return nil, fmt.Errorf("netsuite custom query cannot be empty")
 	}
-	return s.readSuiteQL(ctx, query, opts)
+	return s.readSuiteQL(ctx, query, "", opts)
 }
 
-func (s *NetSuiteSource) readSuiteQL(ctx context.Context, query string, opts source.ReadOptions) (<-chan source.RecordBatchResult, error) {
+func (s *NetSuiteSource) readSuiteQL(ctx context.Context, query, fallbackQuery string, opts source.ReadOptions) (<-chan source.RecordBatchResult, error) {
 	if s.client == nil {
 		return nil, fmt.Errorf("netsuite source is not connected")
 	}
@@ -137,6 +139,17 @@ func (s *NetSuiteSource) readSuiteQL(ctx context.Context, query string, opts sou
 
 			resp, err := s.client.SuiteQL(ctx, query, pageSize, offset)
 			if err != nil {
+				// Ordering is best-effort: if NetSuite rejects the auto-added
+				// ORDER BY (e.g. the table exposes no such column), retry once
+				// without it so the load still succeeds. Pagination may then be
+				// inconsistent across pages, so warn the user.
+				if fallbackQuery != "" && offset == 0 && isInvalidQueryError(err) {
+					fmt.Printf("Warning: netsuite rejected the ordered query; retrying without ORDER BY. Multi-page results may be inconsistent—pass --primary-key to set a sortable key.\n")
+					config.Debug("[NETSUITE] Falling back to unordered query: %s", fallbackQuery)
+					query = fallbackQuery
+					fallbackQuery = ""
+					continue
+				}
 				results <- source.RecordBatchResult{Err: err}
 				return
 			}
@@ -267,8 +280,14 @@ func (c uriConfig) authProvider() (AuthProvider, []interface{ Close() error }, e
 	return provider, []interface{ Close() error }{provider}, nil
 }
 
-func buildSuiteQL(tableName string, opts source.ReadOptions) string {
-	query := fmt.Sprintf("SELECT * FROM %s", strings.TrimSpace(tableName))
+// buildSuiteQL returns the query to run and an unordered fallback. Offset/limit
+// pagination is only stable with a deterministic ORDER BY, so the primary query
+// sorts by the incremental key first, then the primary keys as a unique
+// tie-breaker. The fallback drops the ORDER BY (but keeps any WHERE filter) so a
+// table without a sortable key can still be loaded; it is empty when no ordering
+// was added.
+func buildSuiteQL(tableName string, primaryKeys []string, opts source.ReadOptions) (string, string) {
+	base := fmt.Sprintf("SELECT * FROM %s", strings.TrimSpace(tableName))
 
 	var conditions []string
 	if opts.IncrementalKey != "" {
@@ -280,11 +299,40 @@ func buildSuiteQL(tableName string, opts source.ReadOptions) string {
 		}
 	}
 	if len(conditions) > 0 {
-		query += " WHERE " + strings.Join(conditions, " AND ")
-		query += " ORDER BY " + opts.IncrementalKey + " ASC"
+		base += " WHERE " + strings.Join(conditions, " AND ")
+	}
+
+	orderBy := buildOrderBy(opts.IncrementalKey, primaryKeys)
+	if orderBy == "" {
+		return base, ""
+	}
+	return base + " ORDER BY " + orderBy, base
+}
+
+func isInvalidQueryError(err error) bool {
+	var apiErr *SuiteQLError
+	return errors.As(err, &apiErr) && apiErr.StatusCode == http.StatusBadRequest
+}
+
+func buildOrderBy(incrementalKey string, primaryKeys []string) string {
+	seen := make(map[string]bool)
+	var columns []string
+	add := func(col string) {
+		col = strings.TrimSpace(col)
+		key := strings.ToLower(col)
+		if col == "" || seen[key] {
+			return
+		}
+		seen[key] = true
+		columns = append(columns, col+" ASC")
+	}
+
+	add(incrementalKey)
+	for _, pk := range primaryKeys {
+		add(pk)
 	}
 
-	return query
+	return strings.Join(columns, ", ")
 }
 
 func suiteQLTimestampLiteral(t time.Time) string {

pkg/source/netsuite/netsuite_test.go

@@ -78,13 +78,32 @@ func TestBuildSuiteQL(t *testing.T) {
 	start := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
 	end := time.Date(2026, 1, 3, 3, 4, 5, 0, time.UTC)
 
-	got := buildSuiteQL("transaction", source.ReadOptions{
+	got, fallback := buildSuiteQL("transaction", []string{"id"}, source.ReadOptions{
 		IncrementalKey: "lastmodifieddate",
 		IntervalStart:  &start,
 		IntervalEnd:    &end,
 	})
 
-	assert.Equal(t, `SELECT * FROM transaction WHERE lastmodifieddate >= TO_TIMESTAMP_TZ('2026-01-02T03:04:05.000 +00:00', 'YYYY-MM-DD"T"HH24:MI:SS.FF TZH:TZM') AND lastmodifieddate < TO_TIMESTAMP_TZ('2026-01-03T03:04:05.000 +00:00', 'YYYY-MM-DD"T"HH24:MI:SS.FF TZH:TZM') ORDER BY lastmodifieddate ASC`, got)
+	where := `SELECT * FROM transaction WHERE lastmodifieddate >= TO_TIMESTAMP_TZ('2026-01-02T03:04:05.000 +00:00', 'YYYY-MM-DD"T"HH24:MI:SS.FF TZH:TZM') AND lastmodifieddate < TO_TIMESTAMP_TZ('2026-01-03T03:04:05.000 +00:00', 'YYYY-MM-DD"T"HH24:MI:SS.FF TZH:TZM')`
+	assert.Equal(t, where+" ORDER BY lastmodifieddate ASC, id ASC", got)
+	assert.Equal(t, where, fallback)
+}
+
+func TestBuildSuiteQLOrdersByPrimaryKeyWithoutIncrementalKey(t *testing.T) {
+	got, fallback := buildSuiteQL("customer", []string{"id"}, source.ReadOptions{})
+	assert.Equal(t, "SELECT * FROM customer ORDER BY id ASC", got)
+	assert.Equal(t, "SELECT * FROM customer", fallback)
+}
+
+func TestBuildSuiteQLDedupesIncrementalKeyInOrderBy(t *testing.T) {
+	got, _ := buildSuiteQL("customer", []string{"id"}, source.ReadOptions{IncrementalKey: "id"})
+	assert.Equal(t, "SELECT * FROM customer ORDER BY id ASC", got)
+}
+
+func TestBuildSuiteQLWithoutPrimaryKeyHasNoFallback(t *testing.T) {
+	got, fallback := buildSuiteQL("customer", nil, source.ReadOptions{})
+	assert.Equal(t, "SELECT * FROM customer", got)
+	assert.Empty(t, fallback)
 }
 
 func TestNetSuiteSourceGetTable(t *testing.T) {
@@ -115,7 +134,7 @@ func TestNetSuiteSourceReadUsesSuiteQLPagination(t *testing.T) {
 
 		var body map[string]string
 		require.NoError(t, json.NewDecoder(r.Body).Decode(&body))
-		assert.Equal(t, "SELECT * FROM customer", body["q"])
+		assert.Equal(t, "SELECT * FROM customer ORDER BY id ASC", body["q"])
 
 		w.Header().Set("Content-Type", "application/json")
 		call := requestCount.Add(1)
@@ -194,7 +213,7 @@ func TestNetSuiteSourceReadHonorsLimit(t *testing.T) {
 	}
 	defer func() { _ = s.Close(context.Background()) }()
 
-	ch, err := s.readTable(context.Background(), "customer", source.ReadOptions{Limit: 2})
+	ch, err := s.readTable(context.Background(), "customer", []string{"id"}, source.ReadOptions{Limit: 2})
 	require.NoError(t, err)
 
 	batches := collectResults(t, ch)
@@ -203,6 +222,57 @@ func TestNetSuiteSourceReadHonorsLimit(t *testing.T) {
 	assert.Equal(t, int32(1), requestCount.Load())
 }
 
+func TestNetSuiteSourceReadFallsBackWhenOrderByRejected(t *testing.T) {
+	var orderedAttempts atomic.Int32
+	var unorderedRequests atomic.Int32
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var body map[string]string
+		require.NoError(t, json.NewDecoder(r.Body).Decode(&body))
+
+		w.Header().Set("Content-Type", "application/json")
+		switch body["q"] {
+		case "SELECT * FROM customrecord ORDER BY id ASC":
+			orderedAttempts.Add(1)
+			w.WriteHeader(http.StatusBadRequest)
+			_ = json.NewEncoder(w).Encode(map[string]interface{}{
+				"o:errorDetails": []map[string]interface{}{
+					{"detail": "Search error occurred: Field 'id' was not found."},
+				},
+			})
+		case "SELECT * FROM customrecord":
+			unorderedRequests.Add(1)
+			_ = json.NewEncoder(w).Encode(map[string]interface{}{
+				"count":        1,
+				"offset":       0,
+				"hasMore":      false,
+				"totalResults": 1,
+				"items": []map[string]interface{}{
+					{"name": "row-1"},
+				},
+			})
+		default:
+			t.Errorf("unexpected query: %q", body["q"])
+			w.WriteHeader(http.StatusInternalServerError)
+		}
+	}))
+	defer server.Close()
+
+	s := &NetSuiteSource{
+		client: NewClient(server.URL, NewStaticTokenProvider("test-token")),
+	}
+	defer func() { _ = s.Close(context.Background()) }()
+
+	ch, err := s.readTable(context.Background(), "customrecord", []string{"id"}, source.ReadOptions{})
+	require.NoError(t, err)
+
+	batches := collectResults(t, ch)
+	require.Len(t, batches, 1)
+	assert.Equal(t, int64(1), batches[0].Batch.NumRows())
+	assert.Equal(t, int32(1), orderedAttempts.Load())
+	assert.Equal(t, int32(1), unorderedRequests.Load())
+}
+
 func TestExecuteCustomQueryTrimsSemicolon(t *testing.T) {
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		var body map[string]string