fix(hbase): cap the cells of a row and refuse a larger row

cmd/combine.go

@@ -67,7 +67,7 @@ func newCombineCmd(cfg *config) *cobra.Command {
 		},
 	}
 	c.Flags().StringVar(&with, "with", "", "final jq over the bound source results (each spec's results bound to $name), run over a null input")
-	c.Flags().BoolVar(&cfg.unbounded, "unbounded", false, "permit a spec that loads a whole keyspace into memory (also materializes a .[]-rooted filter instead of streaming it, and permits a Redis key with more than 1000000 elements)")
+	c.Flags().BoolVar(&cfg.unbounded, "unbounded", false, "permit a spec that loads a whole keyspace into memory (also materializes a .[]-rooted filter instead of streaming it, and permits a Redis key with more than 1000000 elements or an HBase row with more than 1000000 cells in one column family)")
 	c.Flags().BoolVar(&cfg.noCompile, "no-compile", false, "disable server-side predicate pushdown; run each spec's full .[]|select(...) filter client-side (results are unchanged either way)")
 	c.Flags().BoolVar(&cfg.explain, "explain", false, "print the formatted query plan (pretty jq, per-source filters, and the backend calls) and exit without connecting or executing")
 	c.Flags().StringVar(&cfg.insert, "insert", "", "write the combined results into this destination `source` instead of rendering (needs --key or --key-field)")

cmd/diff.go

@@ -143,7 +143,7 @@ func newDiffCmd(cfg *config) *cobra.Command {
 	c.Flags().IntVar(&sample, "sample", 1000, "max items sampled per side for --schema (0 = all)")
 	c.Flags().StringVar(&filter, "filter", "", "jq filter rooted at .[] scoping both sides; a spec's own source=<jq> overrides it for that side")
 	c.Flags().BoolVar(&setArrays, "set-arrays", false, "compare arrays order-insensitively as multisets (duplicates counted)")
-	c.Flags().BoolVar(&cfg.unbounded, "unbounded", false, "permit a Redis key with more than 1000000 elements to be loaded into memory")
+	c.Flags().BoolVar(&cfg.unbounded, "unbounded", false, "permit a Redis key with more than 1000000 elements, or an HBase row with more than 1000000 cells in one column family, to be loaded into memory")
 	c.Flags().BoolVarP(&jsonOut, "json", "j", false, "emit machine-readable JSON")
 	c.Flags().BoolVarP(&yamlOut, "yaml", "y", false, "emit machine-readable YAML")
 	c.Flags().BoolVar(&patchOut, "patch", false, "emit an RFC 6902 JSON Patch (single layer only; excludes --json/--yaml/--set-arrays)")

cmd/exec.go

@@ -70,7 +70,9 @@ func newExecCmd(cfg *config) *cobra.Command {
 			}
 			defer func() { _ = store.Close() }()
 
-			result, err := query.NewRunner(store).Run(ctx, args)
+			// The CLI exec has no cap: it prints the whole reply of the statement
+			// that the user typed. So a capped read, such as an HBase row, is unbounded.
+			result, err := query.NewRunner(store).Run(query.WithUnbounded(ctx), args)
 			if err != nil {
 				return err
 			}

cmd/exec_unbounded_test.go

@@ -0,0 +1,43 @@
+package cmd
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/zsltg/iq/internal/query"
+)
+
+// unboundedProbeStore records whether the context of a statement is unbounded.
+type unboundedProbeStore struct {
+	fakeInspectStore
+	unbounded *bool
+}
+
+func (s unboundedProbeStore) Query(ctx context.Context, _ []string) (any, error) {
+	*s.unbounded = query.Unbounded(ctx)
+	return []any{}, nil
+}
+
+// TestCLIExecIsUnbounded pins that the CLI exec runs its statement unbounded, so a
+// capped read such as an HBase row comes back whole.
+func TestCLIExecIsUnbounded(t *testing.T) {
+	unbounded := false
+	orig := drivers
+	t.Cleanup(func() { drivers = orig })
+	drivers = append(append([]driver{}, orig...), driver{
+		name: "unbfake", schemes: []string{"unbfake"},
+		open: func(context.Context, *config) (store, error) {
+			return unboundedProbeStore{unbounded: &unbounded}, nil
+		},
+	})
+	c := newSeed()
+	require.NoError(t, c.Add("probe", "unbfake://h"))
+	seedConfig(t, c)
+
+	_, _, err := runSplit(t, "exec", "--src", "probe", "scan", "t")
+
+	require.NoError(t, err)
+	require.True(t, unbounded)
+}

cmd/root.go

@@ -451,7 +451,7 @@ func addRootFlags(root *cobra.Command, cfg *config) {
 	root.PersistentFlags().BoolVar(&cfg.errorTextVerbose, "error.format.text.verbose", true, "for a jq syntax error in text format, show a caret span report")
 	root.PersistentFlags().StringVar(&cfg.pprofMode, "debug.pprof", "", "write a runtime profile of the whole run: cpu, mem, block, mutex, goroutine, thread, or trace")
 	// --unbounded and --no-compile are local to the default jq action.
-	root.Flags().BoolVar(&cfg.unbounded, "unbounded", false, "permit a filter that loads the whole dataset into memory (also materializes a .[]-rooted filter instead of streaming it, and permits a Redis key with more than 1000000 elements)")
+	root.Flags().BoolVar(&cfg.unbounded, "unbounded", false, "permit a filter that loads the whole dataset into memory (also materializes a .[]-rooted filter instead of streaming it, and permits a Redis key with more than 1000000 elements or an HBase row with more than 1000000 cells in one column family)")
 	root.Flags().BoolVar(&cfg.noCompile, "no-compile", false, "disable predicate pushdown; run the full .[]|select(...) filter client-side (pushdown is on by default where the backend supports it, see iq --explain; results are unchanged either way)")
 	root.Flags().BoolVar(&cfg.explain, "explain", false, "print the formatted query plan (pretty jq, nested filters, and the backend calls) and exit without connecting or executing")
 	addRenderFlags(root, cfg)

cmd/schema.go

@@ -94,7 +94,7 @@ func newSchemaCmd(cfg *config) *cobra.Command {
 			return writeStructured(cmd.OutOrStdout(), doc, asYAML)
 		},
 	}
-	c.Flags().BoolVar(&cfg.unbounded, "unbounded", false, "permit a Redis key with more than 1000000 elements to be loaded into memory")
+	c.Flags().BoolVar(&cfg.unbounded, "unbounded", false, "permit a Redis key with more than 1000000 elements, or an HBase row with more than 1000000 cells in one column family, to be loaded into memory")
 	c.Flags().IntVar(&sample, "sample", 1000, "max items sampled (0 = all)")
 	c.Flags().StringVar(&filter, "filter", "", "jq filter scoping which items the shape is inferred from; the spec form source=<jq> sets it per source")
 	c.Flags().BoolVarP(&yamlOut, "yaml", "y", false, "emit YAML instead of JSON (jsonschema only; odcs is always YAML)")

docs/docs/drivers.md

@@ -678,6 +678,10 @@ iq add -n zk 'hbase://z1,z2,z3:2181/?table=ns:events&znode=/hbase-unsecure'
 A row is a **nested object**: `{family: {qualifier: value}}`, so a cell is addressed as
 `.cf.title` in jq. The row key is the map key, not a field of the row.
 
+A read refuses a row with more than 1,000,000 cells in one column family. The error names the row
+key. The server stops the row at that limit, so the read never loads the whole row. Pass `--unbounded`
+to read such a row. The limit applies to `get` and `scan` through the MCP `iq_exec` tool too. The CLI `iq exec` has no limit. A filtered scan that pushes a column test to the server reads a row whole, because the cap could hide the column that the test needs. It still refuses a row over the limit.
+
 ### Value encoding
 
 HBase stores **no types**. Every cell is raw bytes. As a result, the driver presents a value

docs/docs/global-flags.md

@@ -62,4 +62,4 @@ and write families on their own pages.
 `iq combine`, `iq diff`, `iq schema` and `iq data` carry their own copies of
 `--unbounded` or `--explain` where they apply. On `iq diff`, `iq schema`, and the
 copy and move modes (`--insert`, `--typed`), `--unbounded` only permits a Redis key
-above the element limit.
+or an HBase row above the size limit.

docs/docs/query-data.md

@@ -140,7 +140,8 @@ collapses the keyspace into one value (`keys`, `.`, `map(...)`) runs only with
 it (see [Read strategies](how-it-works.md#read-strategies)).
 
 The flag also lets a read load a Redis key with more than 1,000,000 elements. Without it, the read fails
-and names the key, its type, and its size (see [Redis](drivers.md#redis)). The same
+and names the key, its type, and its size (see [Redis](drivers.md#redis)). It does the same for an
+HBase row with more than 1,000,000 cells in one column family (see [HBase](drivers.md#hbase)). The same
 flag does this for `iq diff`, `iq schema`, and the copy and move modes.
 
 `iq combine` and `iq data` carry their own copies of the flag where they apply.

docs/docs/sources.md

@@ -313,7 +313,7 @@ Exits non-zero when the sources differ and zero when they match
 | | `--patch` | ✗ | emit an RFC 6902 JSON Patch[^1] that transforms the left source into the right (single layer only). For `--data`, the pointers read `/<key>/<field>` over the whole keyspace map. `--patch` excludes `--json`/`--yaml`/`--set-arrays` |
 | | `--schema` | ✗ | diff an inferred field/type shape (cross-driver allowed) based on a sample (use `--sample` to change the sample size) |
 | | `--sample <int>` | `1000` | max items sampled per side for --schema (0 = all) |
-| | `--unbounded` | ✗ | permit a Redis key with more than 1,000,000 elements to be loaded into memory |
+| | `--unbounded` | ✗ | permit a Redis key with more than 1,000,000 elements, or an HBase row with more than 1,000,000 cells in one column family, to be loaded into memory |
 | | `--stats` | ✗ | diff native introspection trees (same driver only). Use with `--section` to narrow it down |
 | | `--section <strings>` | full set | introspection section(s) for `--stats` (comma-separated or repeatable) |
 | | `--set-arrays` | ✗ | compare arrays order-insensitively as multisets (duplicates counted). Membership deltas are reported at the array's own path with no index segment. A pure reorder becomes no difference |
@@ -401,7 +401,7 @@ refuse it.
 | | `--filter <string>` | none | jq filter scoping which items the shape is inferred from. The spec form `source=<jq>` sets it per source |
 | | `--format <string>` | `jsonschema` | picks the contract dialect the shape projects into: JSON Schema draft 2020-12[^4] (`jsonschema`) or Open Data Contract Standard v3.2.0[^5] (`odcs`) |
 | | `--sample <int>` | `1000` | max items sampled (0 = all) |
-| | `--unbounded` | ✗ | permit a Redis key with more than 1,000,000 elements to be loaded into memory |
+| | `--unbounded` | ✗ | permit a Redis key with more than 1,000,000 elements, or an HBase row with more than 1,000,000 cells in one column family, to be loaded into memory |
 | `-y` | `--yaml` | ✗ | emit YAML instead of JSON (`jsonschema` only, `odcs` is always YAML) |
 
 ```sh { title='Schema of the active source' }

docs/man/iq.1

@@ -49,7 +49,7 @@ skip the file:// cache's per\-page key index (still caches the decode; a bounded
 disable predicate pushdown; run the full .[]|select(...) filter client\-side (pushdown is on by default where the backend supports it, see iq \-\-explain; results are unchanged either way)
 .TP
 \fB\-\-unbounded\fR
-permit a filter that loads the whole dataset into memory (also materializes a .[]\-rooted filter instead of streaming it, and permits a Redis key with more than 1000000 elements)
+permit a filter that loads the whole dataset into memory (also materializes a .[]\-rooted filter instead of streaming it, and permits a Redis key with more than 1000000 elements or an HBase row with more than 1000000 cells in one column family)
 .TP
 \fB\-\-compact\fR
 collapse pretty json / jsona output to single\-line (no\-op for jsonl, raw, yaml, gron, grona, parquet)
@@ -402,7 +402,7 @@ empty the destination before writing, with confirmation (\-\-insert)
 native type stamped on each written value, e.g. hash, list, json (\-\-insert)
 .TP
 \fB\-\-unbounded\fR
-permit a spec that loads a whole keyspace into memory (also materializes a .[]\-rooted filter instead of streaming it, and permits a Redis key with more than 1000000 elements)
+permit a spec that loads a whole keyspace into memory (also materializes a .[]\-rooted filter instead of streaming it, and permits a Redis key with more than 1000000 elements or an HBase row with more than 1000000 cells in one column family)
 .TP
 \fB\-\-with\fR
 final jq over the bound source results (each spec's results bound to $name), run over a null input
@@ -582,7 +582,7 @@ compare arrays order\-insensitively as multisets (duplicates counted)
 diff native introspection trees (same driver only)
 .TP
 \fB\-\-unbounded\fR
-permit a Redis key with more than 1000000 elements to be loaded into memory
+permit a Redis key with more than 1000000 elements, or an HBase row with more than 1000000 cells in one column family, to be loaded into memory
 .TP
 \fB\-y\fR, \fB\-\-yaml\fR
 emit machine\-readable YAML
@@ -762,7 +762,7 @@ contract format: jsonschema or odcs (default jsonschema)
 max items sampled (0 = all) (default 1000)
 .TP
 \fB\-\-unbounded\fR
-permit a Redis key with more than 1000000 elements to be loaded into memory
+permit a Redis key with more than 1000000 elements, or an HBase row with more than 1000000 cells in one column family, to be loaded into memory
 .TP
 \fB\-y\fR, \fB\-\-yaml\fR
 emit YAML instead of JSON (jsonschema only; odcs is always YAML)

drivers/hbase/fake_test.go

@@ -8,6 +8,8 @@ import (
 	"sync/atomic"
 	"testing"
 
+	"github.com/zsltg/iq/internal/query"
+
 	"github.com/stretchr/testify/require"
 	"github.com/tsuna/gohbase/hrpc"
 	"github.com/tsuna/gohbase/pb"
@@ -363,11 +365,12 @@ func tableName(namespace, qualifier string) *pb.TableName {
 // size so paging is exercised by a handful of rows.
 func newFakeStore(client hbaseClient, admin hbaseAdmin, table string, types typeMap, rowkey colType) *Store {
 	return &Store{
-		client:     client,
-		admin:      admin,
-		table:      table,
-		types:      types,
-		rowkeyType: rowkey,
-		pageSize:   2,
+		client:      client,
+		admin:       admin,
+		table:       table,
+		types:       types,
+		rowkeyType:  rowkey,
+		pageSize:    2,
+		maxRowCells: query.MaxValueElements,
 	}
 }

drivers/hbase/filter.go

@@ -17,10 +17,10 @@ func (s *Store) ScanFiltered(ctx context.Context, pred predicate.Node, fn func(b
 	if s.table == "" {
 		return errNoTable
 	}
-	opts := []func(hrpc.Call) error{hrpc.MaxVersions(1)}
+	opts := s.readOptions(ctx)
 	if f, ok := toFilter(s.types, pred); ok {
 		s.traceOp("scan %s (filtered)", s.table)
-		opts = append(opts, hrpc.Filters(f))
+		opts = filteredOptions(f)
 	} else {
 		s.traceOp("scan %s", s.table)
 	}
@@ -31,6 +31,15 @@ func (s *Store) ScanFiltered(ctx context.Context, pred predicate.Node, fn func(b
 	return s.pageScan(ctx, req, fn)
 }
 
+// filteredOptions returns the options of a scan with the server filter f. It sends
+// no cell cap. The filter drops a row that misses its column, and a cap could stop
+// the read of a column family before the server reaches that column, so a row over
+// the cap would be dropped without an error. Without the cap the server returns the
+// whole row, and readRow still refuses a row over the limit.
+func filteredOptions(f filter.Filter) []func(hrpc.Call) error {
+	return []func(hrpc.Call) error{hrpc.MaxVersions(1), hrpc.Filters(f)}
+}
+
 // toFilter translates the pushable part of a neutral predicate into an HBase filter.
 // Only column equality (Eq on a family:qualifier path) and an And of such
 // equalities are pushed. An And pushes the conjuncts that compile and drops the

drivers/hbase/hbase.go

@@ -76,7 +76,10 @@ type Store struct {
 	types      typeMap
 	rowkeyType colType
 	pageSize   int
-	trace      io.Writer
+	// maxRowCells is the most cells that one column family of a row can hold and
+	// still be read without the unbounded opt-in. A test lowers it.
+	maxRowCells uint32
+	trace       io.Writer
 }
 
 // Open connects to the HBase cluster named by an hbase:// URL. A bounded
@@ -105,13 +108,14 @@ func Open(ctx context.Context, rawURL, address string, trace io.Writer, _ numfmt
 		opts = append(opts, gohbase.ZookeeperTimeout(d), gohbase.RegionLookupTimeout(d), gohbase.RegionReadTimeout(d))
 	}
 	st := &Store{
-		client:     gohbase.NewClient(cc.zkquorum, opts...),
-		admin:      gohbase.NewAdminClient(cc.zkquorum, opts...),
-		table:      cc.table,
-		types:      cc.types,
-		rowkeyType: cc.rowkeyType,
-		pageSize:   scanBatch,
-		trace:      trace,
+		client:      gohbase.NewClient(cc.zkquorum, opts...),
+		admin:       gohbase.NewAdminClient(cc.zkquorum, opts...),
+		table:       cc.table,
+		types:       cc.types,
+		rowkeyType:  cc.rowkeyType,
+		pageSize:    scanBatch,
+		maxRowCells: query.MaxValueElements,
+		trace:       trace,
 	}
 	if err := st.probe(ctx); err != nil {
 		_ = st.Close()
@@ -262,27 +266,42 @@ func (s *Store) Get(ctx context.Context, keys []string) (map[string]any, error)
 	}
 	s.traceOp("get %s rows=%d", s.table, len(keys))
 	for _, k := range keys {
-		rk, err := encodeRowKey(s.rowkeyType, k)
+		row, found, err := s.getRow(ctx, k)
 		if err != nil {
 			return nil, err
 		}
-		req, err := hrpc.NewGet(ctx, []byte(s.table), rk, hrpc.MaxVersions(1))
-		if err != nil {
-			return nil, fmt.Errorf("hbase get: %w", err)
-		}
-		res, err := s.client.Get(req)
-		if err != nil {
-			return nil, fmt.Errorf("hbase get: %w", err)
+		if found {
+			out[k] = row
 		}
-		// A key with no cells is left out, matching the KV contract.
-		if len(res.Cells) == 0 {
-			continue
-		}
-		out[k] = s.keyedRow(ctx, k, res.Cells)
 	}
 	return out, nil
 }
 
+// getRow reads one row by key. A key with no cells is not found, matching the KV
+// contract.
+func (s *Store) getRow(ctx context.Context, k string) (any, bool, error) {
+	rk, err := encodeRowKey(s.rowkeyType, k)
+	if err != nil {
+		return nil, false, err
+	}
+	req, err := hrpc.NewGet(ctx, []byte(s.table), rk, s.readOptions(ctx)...)
+	if err != nil {
+		return nil, false, fmt.Errorf("hbase get: %w", err)
+	}
+	res, err := s.client.Get(req)
+	if err != nil {
+		return nil, false, fmt.Errorf("hbase get: %w", err)
+	}
+	if len(res.Cells) == 0 {
+		return nil, false, nil
+	}
+	row, err := s.readRow(ctx, k, res.Cells)
+	if err != nil {
+		return nil, false, err
+	}
+	return row, true, nil
+}
+
 // ScanBatches streams the whole table. It hands the caller each page of
 // {key: row} as the region-server cursor yields it. A streaming caller keeps only
 // one page in memory. ctx bounds the scan. It stops at the first error from fn or
@@ -292,7 +311,7 @@ func (s *Store) ScanBatches(ctx context.Context, fn func(batch map[string]any) e
 		return errNoTable
 	}
 	s.traceOp("scan %s", s.table)
-	req, err := hrpc.NewScanStr(ctx, s.table, hrpc.MaxVersions(1))
+	req, err := hrpc.NewScanStr(ctx, s.table, s.readOptions(ctx)...)
 	if err != nil {
 		return fmt.Errorf("hbase scan: %w", err)
 	}
@@ -308,7 +327,11 @@ func (s *Store) pageScan(ctx context.Context, req *hrpc.Scan, fn func(batch map[
 	page := make(map[string]any, s.pageSize)
 	err := eachRow(scanner, "hbase scan", func(res *hrpc.Result) (bool, error) {
 		key := rowKeyString(s.rowkeyType, res.Cells[0].Row)
-		page[key] = s.keyedRow(ctx, key, res.Cells)
+		row, err := s.readRow(ctx, key, res.Cells)
+		if err != nil {
+			return false, err
+		}
+		page[key] = row
 		if len(page) < s.pageSize {
 			return false, nil
 		}
@@ -395,7 +418,7 @@ func (s *Store) execGet(ctx context.Context, args []string) (any, error) {
 	if err != nil {
 		return nil, err
 	}
-	req, err := hrpc.NewGet(ctx, []byte(table), rk, hrpc.MaxVersions(1))
+	req, err := hrpc.NewGet(ctx, []byte(table), rk, s.readOptions(ctx)...)
 	if err != nil {
 		return nil, fmt.Errorf("hbase get: %w", err)
 	}
@@ -406,7 +429,7 @@ func (s *Store) execGet(ctx context.Context, args []string) (any, error) {
 	if len(res.Cells) == 0 {
 		return nil, nil
 	}
-	return s.keyedRow(ctx, key, res.Cells), nil
+	return s.readRow(ctx, key, res.Cells)
 }
 
 // execScan runs `scan <table> [limit]`, returning up to limit rows as {rowkey: row}.
@@ -419,14 +442,18 @@ func (s *Store) execScan(ctx context.Context, args []string) (any, error) {
 	if err != nil {
 		return nil, err
 	}
-	req, err := hrpc.NewScanStr(ctx, args[0], hrpc.MaxVersions(1))
+	req, err := hrpc.NewScanStr(ctx, args[0], s.readOptions(ctx)...)
 	if err != nil {
 		return nil, fmt.Errorf("hbase scan: %w", err)
 	}
 	rows := map[string]any{}
 	err = eachRow(s.client.Scan(req), "hbase scan", func(res *hrpc.Result) (bool, error) {
 		key := rowKeyString(s.rowkeyType, res.Cells[0].Row)
-		rows[key] = s.keyedRow(ctx, key, res.Cells)
+		row, err := s.readRow(ctx, key, res.Cells)
+		if err != nil {
+			return false, err
+		}
+		rows[key] = row
 		return limit > 0 && len(rows) >= limit, nil
 	})
 	if err != nil {