Serve canonical constraint records the extension will read

vgi.v3 replaced the descriptor constraint fields with provider-owned
ConstraintInfo records, but Java left a worker only raw nested byte[]
plumbing, negotiated the read profile only for the main catalog through a
hand-written selector, and never checked the rules the extension enforces
at ATTACH, so a mistake surfaced as a failed ATTACH rather than a worker error.

- ConstraintRecord: typed record with a builder (NOT_NULL incl. nested
  paths, PRIMARY_KEY, UNIQUE, CHECK, FOREIGN_KEY), explicit enforcement and
  validation (no default attestation), encode()/decode(). CatalogTable gains
  Builder#constraints(List<ConstraintRecord>, rev) and withConstraintRecords.
- Every catalog path (main, auxiliary, hosted CatalogInterface) advertises
  vgi.constraint.records.read.v1 + vgi.expression.duckdb.scalar.v1 when its
  tables carry records and selects what the client's offer allows (scalar
  only with records), merged into any explicit catalogCapabilities decision.
  Hosted catalogs get CatalogInterface#catalogReadProfiles (default: scan).
- TableInfoSerializer refuses, naming table and column path: records that
  are not ENFORCED + ALL_ROWS, records owned by another table, duplicate ids,
  and Arrow nullability that disagrees with NOT_NULL/PRIMARY_KEY (every
  table, recursing into ordinary STRUCTs, skipping the row-id column).
- Legacy descriptor constraint fields still fail, now naming the replacement.
- The example worker drops its hand-written negotiation.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: rustyconover

vgi-example-worker/src/main/java/farm/query/vgi/example/FixtureConstraints.java

@@ -20,22 +20,6 @@ private static byte[] ref(String kind, String name, String owner) {
         return RecordCodec.serializeToBytes(new CatalogObjectRef(kind, List.of("data"), name, owner));
     }
 
-    static farm.query.vgi.Worker.CatalogCapabilitySelection negotiate(CatalogAttachRequest request) {
-        if (!"example".equals(request.name())) return new farm.query.vgi.Worker.CatalogCapabilitySelection(null, null);
-        var read = List.of("vgi.constraint.records.read.v1", "vgi.expression.duckdb.scalar.v1");
-        var advertised = new CatalogCapabilities(read, List.of(), List.of(), List.of());
-        var client = request.client_capabilities() == null ? null
-                : RecordCodec.deserializeFromBytes(request.client_capabilities(), ClientCapabilities.class);
-        var offer = client == null || client.catalog() == null
-                ? new CatalogCapabilities(List.of(), List.of(), List.of(), List.of())
-                : RecordCodec.deserializeFromBytes(client.catalog(), CatalogCapabilities.class);
-        var selected = new CatalogCapabilities(read.stream().filter(offer.read()::contains)
-                .filter(profile -> profile.equals(read.getFirst()) || offer.read().contains(read.getFirst())).sorted().toList(),
-                List.of(), List.of(), List.of());
-        return new farm.query.vgi.Worker.CatalogCapabilitySelection(RecordCodec.serializeToBytes(advertised),
-                RecordCodec.serializeToBytes(selected));
-    }
-
     static List<byte[]> snapshot(String table, org.apache.arrow.vector.types.pojo.Field[] fields) {
         if (table.startsWith("versioned_constraints_v")) {
             int version = Integer.parseInt(table.substring(table.length() - 1));

vgi-example-worker/src/main/java/farm/query/vgi/example/Main.java

@@ -366,7 +366,6 @@ private static boolean isTruthy(String v) {
      */
     public static Worker buildWorker(String catalogName, String implVer, String dataSpec) {
         Worker w = Worker.builder()
-                .catalogCapabilities(FixtureConstraints::negotiate)
                 .catalogName(catalogName)
                 .implementationVersion(implVer)
                 .dataVersionSpec(dataSpec)

vgi-example-worker/src/test/java/farm/query/vgi/example/FixtureConstraintsTest.java

@@ -11,20 +11,28 @@
 import farm.query.vgi.protocol.*;
 
 class FixtureConstraintsTest {
-    @Test void readsOnlySelectOfferedProfilesAndTheirPrerequisite() {
+    @Test void exampleCatalogNegotiatesConstraintReadsWithoutProviderCode() {
         String read = "vgi.constraint.records.read.v1";
         String scalar = "vgi.expression.duckdb.scalar.v1";
+        var worker = Main.buildWorker("example", null, null);
+        var svc = new farm.query.vgi.internal.VgiServiceImpl(worker, worker.scalars(), worker.tables(),
+                worker.tableInOuts(), worker.aggregates());
         for (var offered : List.of(List.<String>of(), List.of(scalar), List.of(read), List.of(read, scalar))) {
             var offer = new CatalogCapabilities(offered, List.of(), List.of("vgi.constraint.enforce.v1"), List.of());
             var client = new ClientCapabilities("fixture-test", List.of(), List.of(), true, List.of(), RecordCodec.serializeToBytes(offer));
-            var request = new CatalogAttachRequest("example", null, null, null, RecordCodec.serializeToBytes(client));
-            var selected = RecordCodec.deserializeFromBytes(FixtureConstraints.negotiate(request).selected(), CatalogCapabilities.class);
+            var result = svc.catalog_attach(new CatalogAttachRequest("example", null, null, null, RecordCodec.serializeToBytes(client)), null);
+            var advertised = RecordCodec.deserializeFromBytes(result.catalog_capabilities(), CatalogCapabilities.class);
+            var selected = RecordCodec.deserializeFromBytes(result.selected_capabilities(), CatalogCapabilities.class);
+            assertEquals(List.of(read, scalar), advertised.read());
             assertEquals(offered.contains(read) ? offered : List.of(), selected.read());
             assertTrue(selected.write().isEmpty());
             assertTrue(selected.ddl().isEmpty());
             assertTrue(selected.transaction().isEmpty());
+            // Every served table satisfies the extension's ATTACH-time constraint rules.
+            for (String schema : List.of("main", "data")) {
+                assertDoesNotThrow(() -> svc.catalog_schema_contents_tables(result.attach_opaque_data(), List.of(schema), null, null));
+            }
         }
-        assertNull(FixtureConstraints.negotiate(CatalogAttachRequest.of("other", null, null, null)).selected());
     }
 
     @Test void invalidImmutableDataCannotBecomeTrustedFacts() {

vgi/src/main/java/farm/query/vgi/CatalogInterface.java

@@ -93,6 +93,31 @@ record AttachInfo(@Nullable String defaultSchema, @Nullable String comment, Map<
      */
     AttachInfo attach(byte[] attachId, CatalogAttachRequest request);
 
+    /**
+     * The read profiles this catalog implements for an attach, advertised on the attach result
+     * and selected against the client's offer by the framework. Tables carrying canonical
+     * constraint records ({@link farm.query.vgi.catalog.ConstraintRecord}) are readable only
+     * under {@link farm.query.vgi.catalog.ConstraintRecord#CONSTRAINT_RECORDS_READ}.
+     *
+     * <p>Default: both constraint read profiles when any table of {@link #schemas} carries
+     * records, found by listing every schema's tables right after {@link #attach}. Override
+     * when that listing is expensive, or when tables gain records after attach.
+     *
+     * @param attachId the attach id
+     * @return the read profiles, or empty
+     */
+    default List<String> catalogReadProfiles(byte[] attachId) {
+        for (SchemaInfo s : schemas(attachId)) {
+            for (TableInfo t : tables(attachId, s.path())) {
+                if (!t.constraints().isEmpty()) {
+                    return List.of(farm.query.vgi.catalog.ConstraintRecord.CONSTRAINT_RECORDS_READ,
+                            farm.query.vgi.catalog.ConstraintRecord.SCALAR_EXPRESSION_READ);
+                }
+            }
+        }
+        return List.of();
+    }
+
     /**
      * Release an attach. Default: nothing to release.
      *

vgi/src/main/java/farm/query/vgi/Worker.java

@@ -240,17 +240,40 @@ public Map<String, List<farm.query.vgi.catalog.ScanBranch>> multiBranchTables()
 
     private Worker() {}
 
-    /** Provider-owned capability advertisement and exact selection for the main catalog. */
+    /**
+     * Provider-owned capability advertisement and exact selection for the main catalog.
+     *
+     * @param advertised {@code CatalogCapabilities} IPC bytes the catalog implements, or null
+     * @param selected   {@code CatalogCapabilities} IPC bytes selected for this attach, or null
+     */
     public record CatalogCapabilitySelection(byte[] advertised, byte[] selected) {}
     private java.util.function.Function<farm.query.vgi.protocol.CatalogAttachRequest, CatalogCapabilitySelection>
             catalogCapabilitySelector;
 
+    /**
+     * Sets the main catalog's own capability advertisement and selection, decided per attach.
+     *
+     * <p>Not needed to serve constraint records: every catalog (main, auxiliary and
+     * {@link #registerCatalog hosted}) whose tables carry
+     * {@link farm.query.vgi.catalog.ConstraintRecord}s advertises the constraint read profiles
+     * and selects those the client offers automatically, merged into this decision.
+     *
+     * @param selector the per-attach decision
+     * @return this builder
+     */
     public Worker catalogCapabilities(java.util.function.Function<farm.query.vgi.protocol.CatalogAttachRequest,
             CatalogCapabilitySelection> selector) {
         catalogCapabilitySelector = java.util.Objects.requireNonNull(selector);
         return this;
     }
 
+    /**
+     * The main catalog's own capability decision for an attach (before the framework merges
+     * in the constraint read profiles).
+     *
+     * @param request the attach request
+     * @return the decision; both sides null when no selector is set
+     */
     public CatalogCapabilitySelection selectCatalogCapabilities(farm.query.vgi.protocol.CatalogAttachRequest request) {
         return catalogCapabilitySelector == null ? new CatalogCapabilitySelection(null, null)
                 : catalogCapabilitySelector.apply(request);

vgi/src/main/java/farm/query/vgi/catalog/CatalogTable.java

@@ -415,7 +415,26 @@ public Builder requiredFilters(List<List<List<String>>> groups) {
             return this;
         }
 
-        /** Sets explicit canonical records and their provider-owned revision. */
+        /**
+         * Sets the table's canonical constraint records and their provider-owned revision.
+         * Each record must be owned by this table, and Arrow nullability must agree with the
+         * {@code NOT_NULL}/{@code PRIMARY_KEY} records; both are checked when the table is served.
+         *
+         * @param records  the records, built with {@link ConstraintRecord#builder}
+         * @param revision provider-owned definition revision (non-empty), or {@code null}
+         * @return this builder
+         */
+        public Builder constraints(List<ConstraintRecord> records, byte[] revision) {
+            return canonicalConstraints(ConstraintRecord.encodeAll(records), revision);
+        }
+
+        /**
+         * Sets explicit canonical records, already encoded, and their provider-owned revision.
+         *
+         * @param records  {@code ConstraintInfo} IPC records ({@link ConstraintRecord#encode()})
+         * @param revision provider-owned definition revision, or {@code null}
+         * @return this builder
+         */
         public Builder canonicalConstraints(List<byte[]> records, byte[] revision) {
             this.constraints = List.copyOf(records);
             this.definitionRevision = revision;
@@ -490,7 +509,24 @@ public CatalogTable withConstraints(List<List<Integer>> pk,
                 requiredFilters, constraints, definitionRevision);
     }
 
-    /** Returns a copy with explicit provider-owned canonical records and revision. */
+    /**
+     * Returns a copy with the given canonical constraint records and revision.
+     *
+     * @param records  the records, built with {@link ConstraintRecord#builder}
+     * @param revision provider-owned definition revision (non-empty), or {@code null}
+     * @return a new table carrying the records
+     */
+    public CatalogTable withConstraintRecords(List<ConstraintRecord> records, byte[] revision) {
+        return withCanonicalConstraints(ConstraintRecord.encodeAll(records), revision);
+    }
+
+    /**
+     * Returns a copy with explicit provider-owned canonical records and revision.
+     *
+     * @param records  {@code ConstraintInfo} IPC records ({@link ConstraintRecord#encode()})
+     * @param revision provider-owned definition revision, or {@code null}
+     * @return a new table carrying the records
+     */
     public CatalogTable withCanonicalConstraints(List<byte[]> records, byte[] revision) {
         return new CatalogTable(schema, name, columns, comment, tags,
                 scanFunctionName, scanFunctionPositional, scanFunctionNamed,