Skip to content

Commit 24cafd7

Browse files
rustyconoverclaude
andcommitted
Let scalar, table-in-out and buffering functions declare required secrets
requiredSecrets() existed only on TableFunction and AggregateFunction; the FunctionInfo for scalar, table-in-out and buffering functions always carried an empty required_secrets list, although the extension resolves declared secrets before bind for every kind. Move the hook to FunctionDescriptor so every kind can declare it, and stamp it (with the null-entry check) in all five FunctionInfo builders. Unit tests cover each kind, the null-entry error and serialization. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent ce1b314 commit 24cafd7

3 files changed

Lines changed: 164 additions & 7 deletions

File tree

‎vgi/src/main/java/farm/query/vgi/function/FunctionDescriptor.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,4 +72,20 @@ default FunctionSpec spec() {
7272
* @return the defaults batch, or {@code null} when no parameter defaults
7373
*/
7474
default VectorSchemaRoot parameterDefaultValues() { return null; }
75+
76+
/**
77+
* Secrets the extension resolves before the first bind and sends on it,
78+
* surfaced on the wire as {@code FunctionInfo.required_secrets}. Every
79+
* function kind (scalar, table, table-in-out, buffering, aggregate) can
80+
* declare them; they arrive on the bind/process params of that kind as the
81+
* resolved-secrets blob (parse it with {@code farm.query.vgi.Secrets}). A
82+
* secret that does not exist is simply absent, not an error. Build entries
83+
* with {@code FunctionRequiredSecret.ofType/scoped/named}; a null entry fails
84+
* the catalog listing rather than being dropped. Default: none.
85+
*
86+
* @return the required-secret declarations, or an empty list
87+
*/
88+
default List<farm.query.vgi.protocol.FunctionRequiredSecret> requiredSecrets() {
89+
return List.of();
90+
}
7591
}

‎vgi/src/main/java/farm/query/vgi/internal/VgiServiceImpl.java‎

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -3226,7 +3226,8 @@ private FunctionInfo toScalarFunctionInfo(ScalarFunction fn, String schemaName)
32263226
static FunctionInfo scalarFunctionInfo(ScalarFunction fn, String schemaName) {
32273227
byte[] out = catalogBindOutput(fn.name(),
32283228
() -> fn.onBind(new ScalarBindParams(fn.name(), Arguments.empty(), null, Map.of())));
3229-
return baseFunctionInfo(fn, schemaName, "SCALAR", out, false);
3229+
return withRequiredSecrets(baseFunctionInfo(fn, schemaName, "SCALAR", out, false),
3230+
fn.requiredSecrets());
32303231
}
32313232

32323233
private FunctionInfo toTableFunctionInfo(TableFunction fn, String schemaName) {
@@ -3258,7 +3259,7 @@ static FunctionInfo tableFunctionInfo(TableFunction fn, String schemaName) {
32583259
fn.requiredSecrets());
32593260
}
32603261

3261-
private FunctionInfo toTableInOutFunctionInfo(TableInOutFunction fn, String schemaName) {
3262+
static FunctionInfo toTableInOutFunctionInfo(TableInOutFunction fn, String schemaName) {
32623263
byte[] out = catalogBindOutput(fn.name(),
32633264
() -> fn.onBind(new TableInOutBindParams(fn.name(), Arguments.empty(), null, Map.of())));
32643265
// has_finalize drives the C++ per-substream FINALIZE-phase init after
@@ -3272,7 +3273,7 @@ private FunctionInfo toTableInOutFunctionInfo(TableInOutFunction fn, String sche
32723273
// max_workers (0 = the C++ parallel-by-default per-substream fan-out;
32733274
// 1 would be the A3 serial opt-out AND disqualify the exchange-mode
32743275
// result cache) and input_from_args for blended registrations.
3275-
return new FunctionInfo(
3276+
return withRequiredSecrets(new FunctionInfo(
32763277
base.comment(), base.tags(), base.name(), base.schema_path(), base.function_type(),
32773278
base.arguments(), base.output_schema(), base.parameter_default_values(),
32783279
base.stability(), base.null_handling(), base.argument_monotonicity(),
@@ -3290,7 +3291,7 @@ private FunctionInfo toTableInOutFunctionInfo(TableInOutFunction fn, String sche
32903291
fn instanceof farm.query.vgi.tableinout.RowTransformFunction,
32913292
fn instanceof farm.query.vgi.tableinout.RowTransformFunction rt
32923293
? rt.inputRowsPerCall() : null,
3293-
base.required_settings(), base.required_secrets());
3294+
base.required_settings(), base.required_secrets()), fn.requiredSecrets());
32943295
}
32953296

32963297
private FunctionInfo toAggregateFunctionInfo(AggregateFunction<?> fn, String schemaName) {
@@ -3335,7 +3336,7 @@ private static FunctionInfo withRequiredSecrets(
33353336
base.required_settings(), required);
33363337
}
33373338

3338-
private FunctionInfo toBufferingFunctionInfo(
3339+
static FunctionInfo toBufferingFunctionInfo(
33393340
farm.query.vgi.buffering.TableBufferingFunction fn, String schemaName) {
33403341
byte[] out = catalogBindOutput(fn.name(),
33413342
() -> fn.onBind(new TableInOutBindParams(fn.name(), Arguments.empty(), null, Map.of())));
@@ -3344,7 +3345,7 @@ private FunctionInfo toBufferingFunctionInfo(
33443345
encodeParameterDefaults(fn), /*hasFinalize=*/true, 1, null);
33453346
// Re-stamp the buffering-specific ordering flags (baseFunctionInfo
33463347
// hardcodes them false; they're only meaningful for TableBuffering).
3347-
return new FunctionInfo(
3348+
return withRequiredSecrets(new FunctionInfo(
33483349
base.comment(), base.tags(), base.name(), base.schema_path(), base.function_type(),
33493350
base.arguments(), base.output_schema(), base.parameter_default_values(),
33503351
base.stability(), base.null_handling(), base.argument_monotonicity(),
@@ -3359,7 +3360,7 @@ private FunctionInfo toBufferingFunctionInfo(
33593360
base.supports_window(), base.streaming_partitioned(), base.has_finalize(),
33603361
fn.sourceOrderDependent(), fn.sinkOrderDependent(), fn.requiresInputBatchIndex(),
33613362
base.input_from_args(), base.input_rows_per_call(),
3362-
base.required_settings(), base.required_secrets());
3363+
base.required_settings(), base.required_secrets()), fn.requiredSecrets());
33633364
}
33643365

33653366
private static byte[] bindOutput(BindResponse r) {
Lines changed: 140 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,140 @@
1+
// Copyright 2026 Query Farm LLC - https://query.farm
2+
3+
package farm.query.vgi.internal;
4+
5+
import farm.query.vgi.buffering.TableBufferingCombineParams;
6+
import farm.query.vgi.buffering.TableBufferingFinalizeParams;
7+
import farm.query.vgi.buffering.TableBufferingFunction;
8+
import farm.query.vgi.buffering.TableBufferingProcessParams;
9+
import farm.query.vgi.function.ArgSpec;
10+
import farm.query.vgi.function.FunctionMetadata;
11+
import farm.query.vgi.protocol.BindResponse;
12+
import farm.query.vgi.protocol.FunctionInfo;
13+
import farm.query.vgi.protocol.FunctionRequiredSecret;
14+
import farm.query.vgi.scalar.ScalarBindParams;
15+
import farm.query.vgi.scalar.ScalarFunction;
16+
import farm.query.vgi.scalar.ScalarProcessParams;
17+
import farm.query.vgi.table.TableProducerState;
18+
import farm.query.vgi.tableinout.TableInOutBindParams;
19+
import farm.query.vgi.tableinout.TableInOutExchangeState;
20+
import farm.query.vgi.tableinout.TableInOutFunction;
21+
import farm.query.vgi.tableinout.TableInOutInitParams;
22+
import org.apache.arrow.memory.BufferAllocator;
23+
import org.apache.arrow.vector.VectorSchemaRoot;
24+
import org.junit.jupiter.api.Test;
25+
26+
import java.util.ArrayList;
27+
import java.util.List;
28+
29+
import static org.junit.jupiter.api.Assertions.assertEquals;
30+
import static org.junit.jupiter.api.Assertions.assertThrows;
31+
import static org.junit.jupiter.api.Assertions.assertTrue;
32+
33+
/**
34+
* Every function kind, not only table and aggregate, can declare required
35+
* secrets, and the declaration reaches {@link FunctionInfo#required_secrets()}
36+
* (the extension resolves them before bind for every kind). Before, scalar,
37+
* table-in-out and buffering functions had no hook and always advertised an
38+
* empty list.
39+
*/
40+
class FunctionInfoRequiredSecretsAllKindsTest {
41+
42+
private static final List<FunctionRequiredSecret> DECLARED = List.of(
43+
FunctionRequiredSecret.ofType("vgi_example"),
44+
FunctionRequiredSecret.scoped("s3", "s3://bucket/"),
45+
FunctionRequiredSecret.named("vgi_example", "mine"));
46+
47+
private static List<FunctionRequiredSecret> withNull() {
48+
List<FunctionRequiredSecret> l = new ArrayList<>(DECLARED);
49+
l.add(null);
50+
return l;
51+
}
52+
53+
static class Scalar implements ScalarFunction {
54+
final List<FunctionRequiredSecret> req;
55+
Scalar(List<FunctionRequiredSecret> req) { this.req = req; }
56+
@Override public String name() { return "scalar_secret"; }
57+
@Override public FunctionMetadata metadata() { return FunctionMetadata.describe("doc"); }
58+
@Override public List<ArgSpec> argumentSpecs() { return List.of(); }
59+
@Override public List<FunctionRequiredSecret> requiredSecrets() { return req; }
60+
@Override public BindResponse onBind(ScalarBindParams p) { throw new UnsupportedOperationException(); }
61+
@Override public VectorSchemaRoot process(ScalarProcessParams p, VectorSchemaRoot in, BufferAllocator a) {
62+
throw new UnsupportedOperationException();
63+
}
64+
}
65+
66+
static class InOut implements TableInOutFunction {
67+
final List<FunctionRequiredSecret> req;
68+
InOut(List<FunctionRequiredSecret> req) { this.req = req; }
69+
@Override public String name() { return "tio_secret"; }
70+
@Override public FunctionMetadata metadata() { return FunctionMetadata.describe("doc"); }
71+
@Override public List<ArgSpec> argumentSpecs() { return List.of(); }
72+
@Override public List<FunctionRequiredSecret> requiredSecrets() { return req; }
73+
@Override public BindResponse onBind(TableInOutBindParams p) { throw new UnsupportedOperationException(); }
74+
@Override public TableInOutExchangeState createExchange(TableInOutInitParams p) {
75+
throw new UnsupportedOperationException();
76+
}
77+
}
78+
79+
static class Buffering implements TableBufferingFunction {
80+
final List<FunctionRequiredSecret> req;
81+
Buffering(List<FunctionRequiredSecret> req) { this.req = req; }
82+
@Override public String name() { return "buffering_secret"; }
83+
@Override public FunctionMetadata metadata() { return FunctionMetadata.describe("doc"); }
84+
@Override public List<ArgSpec> argumentSpecs() { return List.of(); }
85+
@Override public List<FunctionRequiredSecret> requiredSecrets() { return req; }
86+
@Override public BindResponse onBind(TableInOutBindParams p) { throw new UnsupportedOperationException(); }
87+
@Override public byte[] process(VectorSchemaRoot b, TableBufferingProcessParams p) {
88+
throw new UnsupportedOperationException();
89+
}
90+
@Override public List<byte[]> combine(List<byte[]> ids, TableBufferingCombineParams p) {
91+
throw new UnsupportedOperationException();
92+
}
93+
@Override public TableProducerState createFinalizeProducer(TableBufferingFinalizeParams p) {
94+
throw new UnsupportedOperationException();
95+
}
96+
}
97+
98+
@Test
99+
void scalarDeclarationReachesFunctionInfo() {
100+
assertEquals(DECLARED, VgiServiceImpl.scalarFunctionInfo(new Scalar(DECLARED), "main").required_secrets());
101+
assertTrue(VgiServiceImpl.scalarFunctionInfo(new Scalar(List.of()), "main").required_secrets().isEmpty());
102+
}
103+
104+
@Test
105+
void tableInOutDeclarationReachesFunctionInfo() {
106+
assertEquals(DECLARED, VgiServiceImpl.toTableInOutFunctionInfo(new InOut(DECLARED), "main").required_secrets());
107+
assertTrue(VgiServiceImpl.toTableInOutFunctionInfo(new InOut(List.of()), "main").required_secrets().isEmpty());
108+
}
109+
110+
@Test
111+
void bufferingDeclarationReachesFunctionInfo() {
112+
assertEquals(DECLARED, VgiServiceImpl.toBufferingFunctionInfo(new Buffering(DECLARED), "main").required_secrets());
113+
assertTrue(VgiServiceImpl.toBufferingFunctionInfo(new Buffering(List.of()), "main").required_secrets().isEmpty());
114+
}
115+
116+
@Test
117+
void nullEntryIsAnErrorForEveryKind() {
118+
assertThrows(IllegalArgumentException.class,
119+
() -> VgiServiceImpl.scalarFunctionInfo(new Scalar(withNull()), "main"));
120+
assertThrows(IllegalArgumentException.class,
121+
() -> VgiServiceImpl.toTableInOutFunctionInfo(new InOut(withNull()), "main"));
122+
assertThrows(IllegalArgumentException.class,
123+
() -> VgiServiceImpl.toBufferingFunctionInfo(new Buffering(withNull()), "main"));
124+
}
125+
126+
@Test
127+
void declarationSurvivesSerializationForEveryKind() {
128+
for (FunctionInfo info : List.of(
129+
VgiServiceImpl.scalarFunctionInfo(new Scalar(DECLARED), "main"),
130+
VgiServiceImpl.toTableInOutFunctionInfo(new InOut(DECLARED), "main"),
131+
VgiServiceImpl.toBufferingFunctionInfo(new Buffering(DECLARED), "main"))) {
132+
try (VectorSchemaRoot encoded = BatchUtil.readSingleBatch(
133+
FunctionInfoSerializer.serialize(info), farm.query.vgirpc.wire.Allocators.root())) {
134+
List<?> items = ((org.apache.arrow.vector.complex.ListVector)
135+
encoded.getVector("required_secrets")).getObject(0);
136+
assertEquals(3, items.size(), info.name());
137+
}
138+
}
139+
}
140+
}

0 commit comments

Comments
 (0)