Skip to content

Commit ff1981c

Browse files
xiangfu0xiangfu0
authored
Use PostgreSQL function volatility for ingestion transforms (#18932)
1 parent 7ecf657 commit ff1981c

26 files changed

Lines changed: 812 additions & 48 deletions

File tree

‎pinot-common/src/main/java/org/apache/pinot/common/function/FunctionInfo.java‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@
1919
package org.apache.pinot.common.function;
2020

2121
import java.lang.reflect.Method;
22+
import java.util.Objects;
23+
import org.apache.pinot.spi.annotations.FunctionVolatility;
2224
import org.apache.pinot.spi.annotations.ScalarFunction;
2325

2426

@@ -27,16 +29,24 @@ public class FunctionInfo {
2729
private final Class<?> _clazz;
2830
private final boolean _nullableParameters;
2931
private final boolean _deterministic;
32+
private final FunctionVolatility _volatility;
3033

3134
public FunctionInfo(Method method, Class<?> clazz, boolean nullableParameters) {
3235
this(method, clazz, nullableParameters, true);
3336
}
3437

3538
public FunctionInfo(Method method, Class<?> clazz, boolean nullableParameters, boolean deterministic) {
39+
this(method, clazz, nullableParameters, deterministic,
40+
deterministic ? resolveVolatility(method, clazz) : FunctionVolatility.VOLATILE);
41+
}
42+
43+
public FunctionInfo(Method method, Class<?> clazz, boolean nullableParameters, boolean deterministic,
44+
FunctionVolatility volatility) {
3645
_method = method;
3746
_clazz = clazz;
3847
_nullableParameters = nullableParameters;
3948
_deterministic = deterministic;
49+
_volatility = Objects.requireNonNull(volatility, "volatility must not be null");
4050
}
4151

4252
public Method getMethod() {
@@ -55,10 +65,39 @@ public boolean isDeterministic() {
5565
return _deterministic;
5666
}
5767

68+
public FunctionVolatility getVolatility() {
69+
return _volatility;
70+
}
71+
5872
public static FunctionInfo fromMethod(Method method) {
5973
ScalarFunction annotation = method.getAnnotation(ScalarFunction.class);
6074
boolean nullableParameters = annotation != null && annotation.nullableParameters();
6175
boolean deterministic = annotation == null || annotation.isDeterministic();
6276
return new FunctionInfo(method, method.getDeclaringClass(), nullableParameters, deterministic);
6377
}
78+
79+
private static FunctionVolatility resolveVolatility(Method method, Class<?> clazz) {
80+
ScalarFunction methodAnnotation = method.getAnnotation(ScalarFunction.class);
81+
ScalarFunction classAnnotation = clazz.getAnnotation(ScalarFunction.class);
82+
FunctionVolatility methodVolatility = getVolatility(methodAnnotation);
83+
FunctionVolatility classVolatility = getVolatility(classAnnotation);
84+
return mostVolatile(methodVolatility, classVolatility);
85+
}
86+
87+
private static FunctionVolatility getVolatility(ScalarFunction annotation) {
88+
if (annotation == null) {
89+
return FunctionVolatility.IMMUTABLE;
90+
}
91+
return annotation.isDeterministic() ? annotation.volatility() : FunctionVolatility.VOLATILE;
92+
}
93+
94+
private static FunctionVolatility mostVolatile(FunctionVolatility first, FunctionVolatility second) {
95+
if (first == FunctionVolatility.VOLATILE || second == FunctionVolatility.VOLATILE) {
96+
return FunctionVolatility.VOLATILE;
97+
}
98+
if (first == FunctionVolatility.STABLE || second == FunctionVolatility.STABLE) {
99+
return FunctionVolatility.STABLE;
100+
}
101+
return FunctionVolatility.IMMUTABLE;
102+
}
64103
}

‎pinot-common/src/main/java/org/apache/pinot/common/function/FunctionRegistry.java‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -128,9 +128,7 @@ private FunctionRegistry() {
128128
}
129129
ScalarFunction scalarFunction = method.getAnnotation(ScalarFunction.class);
130130
if (scalarFunction.enabled()) {
131-
FunctionInfo functionInfo =
132-
new FunctionInfo(method, method.getDeclaringClass(), scalarFunction.nullableParameters(),
133-
scalarFunction.isDeterministic());
131+
FunctionInfo functionInfo = FunctionInfo.fromMethod(method);
134132
int numArguments = scalarFunction.isVarArg() ? VAR_ARG_KEY : method.getParameterCount();
135133
String[] names = scalarFunction.names();
136134
if (names.length == 0) {

‎pinot-common/src/main/java/org/apache/pinot/common/function/PinotScalarFunction.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,11 @@ default FunctionInfo getFunctionInfo(ColumnDataType[] argumentTypes) {
9696
static PinotScalarFunction fromMethod(Method method, boolean isVarArg, boolean supportNullArgs,
9797
@Nullable String... names) {
9898
int numArguments = isVarArg ? FunctionRegistry.VAR_ARG_KEY : method.getParameterCount();
99-
FunctionInfo functionInfo = new FunctionInfo(method, method.getDeclaringClass(), supportNullArgs);
99+
FunctionInfo annotationFunctionInfo = FunctionInfo.fromMethod(method);
100+
// Preserve the historical dynamic UDF compile-time policy while propagating volatility independently.
101+
FunctionInfo functionInfo =
102+
new FunctionInfo(method, method.getDeclaringClass(), supportNullArgs, true,
103+
annotationFunctionInfo.getVolatility());
100104
Map<Integer, FunctionInfo> functionInfoMap = Map.of(numArguments, functionInfo);
101105

102106
List<String> nameList = names != null && names.length > 0

‎pinot-common/src/main/java/org/apache/pinot/common/function/scalar/ArithmeticFunctions.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import java.math.BigDecimal;
2222
import java.math.RoundingMode;
2323
import java.util.concurrent.ThreadLocalRandom;
24+
import org.apache.pinot.spi.annotations.FunctionVolatility;
2425
import org.apache.pinot.spi.annotations.ScalarFunction;
2526

2627

@@ -155,7 +156,7 @@ public static double truncate(double a) {
155156
return Math.signum(a) * Math.floor(Math.abs(a));
156157
}
157158

158-
@ScalarFunction(isDeterministic = false)
159+
@ScalarFunction(isDeterministic = false, volatility = FunctionVolatility.VOLATILE)
159160
public static double rand() {
160161
return ThreadLocalRandom.current().nextDouble();
161162
}

‎pinot-common/src/main/java/org/apache/pinot/common/function/scalar/DateTimeFunctions.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.apache.pinot.common.function.DateTimeUtils;
2828
import org.apache.pinot.common.function.FunctionUtils;
2929
import org.apache.pinot.common.function.TimeZoneKey;
30+
import org.apache.pinot.spi.annotations.FunctionVolatility;
3031
import org.apache.pinot.spi.annotations.ScalarFunction;
3132
import org.apache.pinot.spi.utils.TimeUtils;
3233
import org.joda.time.DateTime;
@@ -573,12 +574,12 @@ public static long[] roundMV(long[] timeValue, long roundToNearest) {
573574
* Return current time as epoch millis
574575
* TODO: Consider changing the return type to Timestamp
575576
*/
576-
@ScalarFunction
577+
@ScalarFunction(volatility = FunctionVolatility.VOLATILE)
577578
public static long now() {
578579
return System.currentTimeMillis();
579580
}
580581

581-
@ScalarFunction
582+
@ScalarFunction(volatility = FunctionVolatility.VOLATILE)
582583
public static long sleep(long millis) {
583584
try {
584585
if (FunctionUtils.isAssertEnabled()) {
@@ -603,13 +604,13 @@ public static long sleep(long millis) {
603604
* "-P6H3M" -- parses as "-6 hours and -3 minutes"
604605
* "-P-6H+3M" -- parses as "+6 hours and -3 minutes"
605606
*/
606-
@ScalarFunction
607+
@ScalarFunction(volatility = FunctionVolatility.VOLATILE)
607608
public static long ago(String periodString) {
608609
Duration period = Duration.parse(periodString);
609610
return System.currentTimeMillis() - period.toMillis();
610611
}
611612

612-
@ScalarFunction
613+
@ScalarFunction(volatility = FunctionVolatility.VOLATILE)
613614
public static long[] agoMV(String[] periodString) {
614615
long[] results = new long[periodString.length];
615616
for (int i = 0; i < periodString.length; i++) {

‎pinot-common/src/main/java/org/apache/pinot/common/function/scalar/InternalFunctions.java‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,12 @@
1818
*/
1919
package org.apache.pinot.common.function.scalar;
2020

21+
import org.apache.pinot.spi.annotations.FunctionVolatility;
2122
import org.apache.pinot.spi.annotations.ScalarFunction;
2223
import org.apache.pinot.spi.query.QueryThreadContext;
2324

2425

26+
@ScalarFunction(enabled = false, volatility = FunctionVolatility.STABLE)
2527
public class InternalFunctions {
2628
private InternalFunctions() {
2729
}
@@ -33,7 +35,7 @@ private InternalFunctions() {
3335
/// input.
3436
///
3537
/// This is mostly useful for test and internal usage
36-
@ScalarFunction
38+
@ScalarFunction(volatility = FunctionVolatility.VOLATILE)
3739
public static String cid(String input) {
3840
return QueryThreadContext.get().getExecutionContext().getCid();
3941
}
@@ -57,7 +59,7 @@ public static long reqId(String input) {
5759
/// input.
5860
///
5961
/// This is mostly useful for test and internal usage and should be close to now()
60-
@ScalarFunction
62+
@ScalarFunction(volatility = FunctionVolatility.VOLATILE)
6163
public static long startTime(String input) {
6264
return QueryThreadContext.get().getExecutionContext().getStartTimeMs();
6365
}
@@ -69,7 +71,7 @@ public static long startTime(String input) {
6971
/// input.
7072
///
7173
/// This is mostly useful for test and internal usage
72-
@ScalarFunction
74+
@ScalarFunction(volatility = FunctionVolatility.VOLATILE)
7375
public static long endTime(String input) {
7476
return QueryThreadContext.get().getExecutionContext().getActiveDeadlineMs();
7577
}
@@ -102,7 +104,7 @@ public static String queryEngine(String input) {
102104
/// input.
103105
///
104106
/// This is mostly useful for test and internal usage
105-
@ScalarFunction
107+
@ScalarFunction(volatility = FunctionVolatility.VOLATILE)
106108
public static int stageId(String input) {
107109
QueryThreadContext.MseWorkerInfo mseWorkerInfo = QueryThreadContext.get().getMseWorkerInfo();
108110
return mseWorkerInfo != null ? mseWorkerInfo.getStageId() : -1;
@@ -116,7 +118,7 @@ public static int stageId(String input) {
116118
/// input.
117119
///
118120
/// This is mostly useful for test and internal usage
119-
@ScalarFunction
121+
@ScalarFunction(volatility = FunctionVolatility.VOLATILE)
120122
public static int workerId(String input) {
121123
QueryThreadContext.MseWorkerInfo mseWorkerInfo = QueryThreadContext.get().getMseWorkerInfo();
122124
return mseWorkerInfo != null ? mseWorkerInfo.getWorkerId() : -1;

‎pinot-common/src/test/java/org/apache/pinot/common/function/FunctionUtilsTest.java‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,14 +26,44 @@
2626
import java.util.List;
2727
import java.util.UUID;
2828
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
29+
import org.apache.pinot.spi.annotations.FunctionVolatility;
30+
import org.apache.pinot.spi.annotations.ScalarFunction;
2931
import org.apache.pinot.spi.utils.PinotDataType;
3032
import org.testng.annotations.Test;
3133

3234
import static org.testng.Assert.assertEquals;
35+
import static org.testng.Assert.assertFalse;
36+
import static org.testng.Assert.assertNotNull;
3337
import static org.testng.Assert.assertNull;
38+
import static org.testng.Assert.assertTrue;
3439

3540

3641
public class FunctionUtilsTest {
42+
/**
43+
* Test fixture for class-level function volatility.
44+
*/
45+
@ScalarFunction(enabled = false, volatility = FunctionVolatility.STABLE)
46+
private static class AnnotatedFunction {
47+
@ScalarFunction(enabled = false)
48+
public static long classAnnotated() {
49+
return 0L;
50+
}
51+
}
52+
53+
/**
54+
* Test fixture for method-level and legacy volatility metadata.
55+
*/
56+
private static class VolatilityAnnotatedFunction {
57+
@ScalarFunction(enabled = false, isDeterministic = false)
58+
public static long legacyVolatile() {
59+
return 0L;
60+
}
61+
62+
@ScalarFunction(enabled = false, volatility = FunctionVolatility.STABLE)
63+
public static long stable() {
64+
return 0L;
65+
}
66+
}
3767

3868
@Test
3969
public void testGetArgumentType() {
@@ -56,6 +86,100 @@ public void testGetArgumentType() {
5686
assertEquals(FunctionUtils.getArgumentType((short) 1), PinotDataType.SHORT);
5787
}
5888

89+
@Test
90+
public void testFunctionVolatilityMetadata() {
91+
FunctionInfo now = FunctionRegistry.lookupFunctionInfo("now", 0);
92+
assertTrue(now.isDeterministic());
93+
assertEquals(now.getVolatility(), FunctionVolatility.VOLATILE);
94+
95+
FunctionInfo ago = FunctionRegistry.lookupFunctionInfo("ago", 1);
96+
assertTrue(ago.isDeterministic());
97+
assertEquals(ago.getVolatility(), FunctionVolatility.VOLATILE);
98+
99+
FunctionInfo agoMv = FunctionRegistry.lookupFunctionInfo("agomv", 1);
100+
assertTrue(agoMv.isDeterministic());
101+
assertEquals(agoMv.getVolatility(), FunctionVolatility.VOLATILE);
102+
103+
FunctionInfo random = FunctionRegistry.lookupFunctionInfo("rand", 0);
104+
assertFalse(random.isDeterministic());
105+
assertEquals(random.getVolatility(), FunctionVolatility.VOLATILE);
106+
107+
FunctionInfo seededRandom = FunctionRegistry.lookupFunctionInfo("rand", 1);
108+
assertTrue(seededRandom.isDeterministic());
109+
assertEquals(seededRandom.getVolatility(), FunctionVolatility.IMMUTABLE);
110+
111+
FunctionInfo sleep = FunctionRegistry.lookupFunctionInfo("sleep", 1);
112+
assertTrue(sleep.isDeterministic());
113+
assertEquals(sleep.getVolatility(), FunctionVolatility.VOLATILE);
114+
115+
FunctionInfo cid = FunctionRegistry.lookupFunctionInfo("cid", 1);
116+
assertTrue(cid.isDeterministic());
117+
assertEquals(cid.getVolatility(), FunctionVolatility.VOLATILE);
118+
119+
FunctionInfo requestId = FunctionRegistry.lookupFunctionInfo("reqid", 1);
120+
assertTrue(requestId.isDeterministic());
121+
assertEquals(requestId.getVolatility(), FunctionVolatility.STABLE);
122+
123+
FunctionInfo startTime = FunctionRegistry.lookupFunctionInfo("starttime", 1);
124+
assertTrue(startTime.isDeterministic());
125+
assertEquals(startTime.getVolatility(), FunctionVolatility.VOLATILE);
126+
127+
FunctionInfo workerId = FunctionRegistry.lookupFunctionInfo("workerid", 1);
128+
assertTrue(workerId.isDeterministic());
129+
assertEquals(workerId.getVolatility(), FunctionVolatility.VOLATILE);
130+
}
131+
132+
@Test
133+
public void testFunctionVolatilityResolution()
134+
throws NoSuchMethodException {
135+
FunctionInfo classAnnotated =
136+
FunctionInfo.fromMethod(AnnotatedFunction.class.getMethod("classAnnotated"));
137+
assertTrue(classAnnotated.isDeterministic());
138+
assertEquals(classAnnotated.getVolatility(), FunctionVolatility.STABLE);
139+
140+
FunctionInfo classAnnotationAwareConstructor =
141+
new FunctionInfo(AnnotatedFunction.class.getMethod("classAnnotated"), AnnotatedFunction.class, false);
142+
assertTrue(classAnnotationAwareConstructor.isDeterministic());
143+
assertEquals(classAnnotationAwareConstructor.getVolatility(), FunctionVolatility.STABLE);
144+
145+
FunctionInfo legacyVolatile =
146+
FunctionInfo.fromMethod(VolatilityAnnotatedFunction.class.getMethod("legacyVolatile"));
147+
assertFalse(legacyVolatile.isDeterministic());
148+
assertEquals(legacyVolatile.getVolatility(), FunctionVolatility.VOLATILE);
149+
150+
FunctionInfo annotationAwareConstructor =
151+
new FunctionInfo(VolatilityAnnotatedFunction.class.getMethod("legacyVolatile"),
152+
VolatilityAnnotatedFunction.class, false, true);
153+
assertTrue(annotationAwareConstructor.isDeterministic());
154+
assertEquals(annotationAwareConstructor.getVolatility(), FunctionVolatility.VOLATILE);
155+
156+
FunctionInfo deterministicOverride =
157+
new FunctionInfo(VolatilityAnnotatedFunction.class.getMethod("stable"),
158+
VolatilityAnnotatedFunction.class, false, false);
159+
assertFalse(deterministicOverride.isDeterministic());
160+
assertEquals(deterministicOverride.getVolatility(), FunctionVolatility.VOLATILE);
161+
162+
FunctionInfo explicitVolatilityOverride = new FunctionInfo(
163+
AnnotatedFunction.class.getMethod("classAnnotated"), AnnotatedFunction.class, false, true,
164+
FunctionVolatility.IMMUTABLE);
165+
assertTrue(explicitVolatilityOverride.isDeterministic());
166+
assertEquals(explicitVolatilityOverride.getVolatility(), FunctionVolatility.IMMUTABLE);
167+
168+
PinotScalarFunction scalarFunction =
169+
PinotScalarFunction.fromMethod(VolatilityAnnotatedFunction.class.getMethod("stable"), false, true);
170+
FunctionInfo dynamicFunctionInfo = scalarFunction.getFunctionInfo(0);
171+
assertNotNull(dynamicFunctionInfo);
172+
assertTrue(dynamicFunctionInfo.isDeterministic());
173+
assertEquals(dynamicFunctionInfo.getVolatility(), FunctionVolatility.STABLE);
174+
175+
PinotScalarFunction legacyDynamicFunction =
176+
PinotScalarFunction.fromMethod(VolatilityAnnotatedFunction.class.getMethod("legacyVolatile"), false, true);
177+
FunctionInfo legacyDynamicFunctionInfo = legacyDynamicFunction.getFunctionInfo(0);
178+
assertNotNull(legacyDynamicFunctionInfo);
179+
assertTrue(legacyDynamicFunctionInfo.isDeterministic());
180+
assertEquals(legacyDynamicFunctionInfo.getVolatility(), FunctionVolatility.VOLATILE);
181+
}
182+
59183
@Test
60184
public void testGetArgumentTypeForVendorTimestampSubclass() {
61185
// Vendor JDBC drivers commonly return Timestamp subclasses (e.g. BigQuery Simba's TimestampTz).

‎pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSchemaRestletResource.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -409,7 +409,8 @@ private void validateSchemaInternal(Schema schema) {
409409
try {
410410
List<TableConfig> tableConfigs = _pinotHelixResourceManager.getTableConfigsForSchema(schema.getSchemaName());
411411
boolean isIgnoreCase = _pinotHelixResourceManager.getTableCache().isIgnoreCase();
412-
SchemaUtils.validate(schema, tableConfigs, isIgnoreCase);
412+
Schema existingSchema = _pinotHelixResourceManager.getSchema(schema.getSchemaName());
413+
SchemaUtils.validate(schema, tableConfigs, isIgnoreCase, existingSchema);
413414
} catch (Exception e) {
414415
throw new ControllerApplicationException(LOGGER,
415416
"Invalid schema: " + schema.getSchemaName() + ". Reason: " + e.getMessage(), Response.Status.BAD_REQUEST, e);

0 commit comments

Comments
 (0)