This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new ff1981c45f4 Use PostgreSQL function volatility for ingestion
transforms (#18932)
ff1981c45f4 is described below
commit ff1981c45f4f8724d558541463f78d970b254ff6
Author: Xiang Fu <[email protected]>
AuthorDate: Tue Jul 28 23:03:16 2026 -0700
Use PostgreSQL function volatility for ingestion transforms (#18932)
---
.../apache/pinot/common/function/FunctionInfo.java | 39 +++++++
.../pinot/common/function/FunctionRegistry.java | 4 +-
.../pinot/common/function/PinotScalarFunction.java | 6 +-
.../function/scalar/ArithmeticFunctions.java | 3 +-
.../common/function/scalar/DateTimeFunctions.java | 9 +-
.../common/function/scalar/InternalFunctions.java | 12 +-
.../pinot/common/function/FunctionUtilsTest.java | 124 +++++++++++++++++++++
.../api/resources/PinotSchemaRestletResource.java | 3 +-
.../api/resources/PinotTableRestletResource.java | 37 +++---
.../api/resources/TableConfigValidationUtils.java | 9 +-
.../api/resources/TableConfigsRestletResource.java | 7 +-
.../api/PinotSchemaRestletResourceTest.java | 41 +++++++
.../api/PinotTableRestletResourceTest.java | 63 +++++++++++
.../api/TableConfigsRestletResourceTest.java | 103 +++++++++++++++++
.../apache/pinot/core/util/SchemaUtilsTest.java | 48 ++++++++
.../tests/custom/CLPEncodingRealtimeTest.java | 2 +-
.../pinot/query/QueryPlannerRuleOptionsTest.java | 7 +-
.../pinot/segment/local/utils/SchemaUtils.java | 43 ++++++-
.../segment/local/utils/TableConfigUtils.java | 115 ++++++++++++++++++-
.../ExpressionTransformerTest.java | 26 +++++
.../segment/local/utils/TableConfigUtilsTest.java | 100 +++++++++++++++++
.../pinot/spi/annotations/FunctionVolatility.java | 37 ++++++
.../pinot/spi/annotations/ScalarFunction.java | 16 ++-
.../fineFoodReviews_realtime_table_config.json | 2 +-
...neFoodReviews_part_0_realtime_table_config.json | 2 +-
...neFoodReviews_part_1_realtime_table_config.json | 2 +-
26 files changed, 812 insertions(+), 48 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/function/FunctionInfo.java
b/pinot-common/src/main/java/org/apache/pinot/common/function/FunctionInfo.java
index 16cd3997bb5..e2236d2736b 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/function/FunctionInfo.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/function/FunctionInfo.java
@@ -19,6 +19,8 @@
package org.apache.pinot.common.function;
import java.lang.reflect.Method;
+import java.util.Objects;
+import org.apache.pinot.spi.annotations.FunctionVolatility;
import org.apache.pinot.spi.annotations.ScalarFunction;
@@ -27,16 +29,24 @@ public class FunctionInfo {
private final Class<?> _clazz;
private final boolean _nullableParameters;
private final boolean _deterministic;
+ private final FunctionVolatility _volatility;
public FunctionInfo(Method method, Class<?> clazz, boolean
nullableParameters) {
this(method, clazz, nullableParameters, true);
}
public FunctionInfo(Method method, Class<?> clazz, boolean
nullableParameters, boolean deterministic) {
+ this(method, clazz, nullableParameters, deterministic,
+ deterministic ? resolveVolatility(method, clazz) :
FunctionVolatility.VOLATILE);
+ }
+
+ public FunctionInfo(Method method, Class<?> clazz, boolean
nullableParameters, boolean deterministic,
+ FunctionVolatility volatility) {
_method = method;
_clazz = clazz;
_nullableParameters = nullableParameters;
_deterministic = deterministic;
+ _volatility = Objects.requireNonNull(volatility, "volatility must not be
null");
}
public Method getMethod() {
@@ -55,10 +65,39 @@ public class FunctionInfo {
return _deterministic;
}
+ public FunctionVolatility getVolatility() {
+ return _volatility;
+ }
+
public static FunctionInfo fromMethod(Method method) {
ScalarFunction annotation = method.getAnnotation(ScalarFunction.class);
boolean nullableParameters = annotation != null &&
annotation.nullableParameters();
boolean deterministic = annotation == null || annotation.isDeterministic();
return new FunctionInfo(method, method.getDeclaringClass(),
nullableParameters, deterministic);
}
+
+ private static FunctionVolatility resolveVolatility(Method method, Class<?>
clazz) {
+ ScalarFunction methodAnnotation =
method.getAnnotation(ScalarFunction.class);
+ ScalarFunction classAnnotation = clazz.getAnnotation(ScalarFunction.class);
+ FunctionVolatility methodVolatility = getVolatility(methodAnnotation);
+ FunctionVolatility classVolatility = getVolatility(classAnnotation);
+ return mostVolatile(methodVolatility, classVolatility);
+ }
+
+ private static FunctionVolatility getVolatility(ScalarFunction annotation) {
+ if (annotation == null) {
+ return FunctionVolatility.IMMUTABLE;
+ }
+ return annotation.isDeterministic() ? annotation.volatility() :
FunctionVolatility.VOLATILE;
+ }
+
+ private static FunctionVolatility mostVolatile(FunctionVolatility first,
FunctionVolatility second) {
+ if (first == FunctionVolatility.VOLATILE || second ==
FunctionVolatility.VOLATILE) {
+ return FunctionVolatility.VOLATILE;
+ }
+ if (first == FunctionVolatility.STABLE || second ==
FunctionVolatility.STABLE) {
+ return FunctionVolatility.STABLE;
+ }
+ return FunctionVolatility.IMMUTABLE;
+ }
}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/function/FunctionRegistry.java
b/pinot-common/src/main/java/org/apache/pinot/common/function/FunctionRegistry.java
index 395c49e8ca8..69fd3f190a8 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/function/FunctionRegistry.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/function/FunctionRegistry.java
@@ -128,9 +128,7 @@ public class FunctionRegistry {
}
ScalarFunction scalarFunction =
method.getAnnotation(ScalarFunction.class);
if (scalarFunction.enabled()) {
- FunctionInfo functionInfo =
- new FunctionInfo(method, method.getDeclaringClass(),
scalarFunction.nullableParameters(),
- scalarFunction.isDeterministic());
+ FunctionInfo functionInfo = FunctionInfo.fromMethod(method);
int numArguments = scalarFunction.isVarArg() ? VAR_ARG_KEY :
method.getParameterCount();
String[] names = scalarFunction.names();
if (names.length == 0) {
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/function/PinotScalarFunction.java
b/pinot-common/src/main/java/org/apache/pinot/common/function/PinotScalarFunction.java
index 1074d6c4f3e..47aac639d55 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/function/PinotScalarFunction.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/function/PinotScalarFunction.java
@@ -96,7 +96,11 @@ public interface PinotScalarFunction {
static PinotScalarFunction fromMethod(Method method, boolean isVarArg,
boolean supportNullArgs,
@Nullable String... names) {
int numArguments = isVarArg ? FunctionRegistry.VAR_ARG_KEY :
method.getParameterCount();
- FunctionInfo functionInfo = new FunctionInfo(method,
method.getDeclaringClass(), supportNullArgs);
+ FunctionInfo annotationFunctionInfo = FunctionInfo.fromMethod(method);
+ // Preserve the historical dynamic UDF compile-time policy while
propagating volatility independently.
+ FunctionInfo functionInfo =
+ new FunctionInfo(method, method.getDeclaringClass(), supportNullArgs,
true,
+ annotationFunctionInfo.getVolatility());
Map<Integer, FunctionInfo> functionInfoMap = Map.of(numArguments,
functionInfo);
List<String> nameList = names != null && names.length > 0
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/ArithmeticFunctions.java
b/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/ArithmeticFunctions.java
index d23b4fb62ed..693cfc94275 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/ArithmeticFunctions.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/ArithmeticFunctions.java
@@ -21,6 +21,7 @@ package org.apache.pinot.common.function.scalar;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.concurrent.ThreadLocalRandom;
+import org.apache.pinot.spi.annotations.FunctionVolatility;
import org.apache.pinot.spi.annotations.ScalarFunction;
@@ -155,7 +156,7 @@ public class ArithmeticFunctions {
return Math.signum(a) * Math.floor(Math.abs(a));
}
- @ScalarFunction(isDeterministic = false)
+ @ScalarFunction(isDeterministic = false, volatility =
FunctionVolatility.VOLATILE)
public static double rand() {
return ThreadLocalRandom.current().nextDouble();
}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/DateTimeFunctions.java
b/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/DateTimeFunctions.java
index 4da99e84cb3..dc498c48c21 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/DateTimeFunctions.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/DateTimeFunctions.java
@@ -27,6 +27,7 @@ import
org.apache.pinot.common.function.DateTimePatternHandler;
import org.apache.pinot.common.function.DateTimeUtils;
import org.apache.pinot.common.function.FunctionUtils;
import org.apache.pinot.common.function.TimeZoneKey;
+import org.apache.pinot.spi.annotations.FunctionVolatility;
import org.apache.pinot.spi.annotations.ScalarFunction;
import org.apache.pinot.spi.utils.TimeUtils;
import org.joda.time.DateTime;
@@ -573,12 +574,12 @@ public class DateTimeFunctions {
* Return current time as epoch millis
* TODO: Consider changing the return type to Timestamp
*/
- @ScalarFunction
+ @ScalarFunction(volatility = FunctionVolatility.VOLATILE)
public static long now() {
return System.currentTimeMillis();
}
- @ScalarFunction
+ @ScalarFunction(volatility = FunctionVolatility.VOLATILE)
public static long sleep(long millis) {
try {
if (FunctionUtils.isAssertEnabled()) {
@@ -603,13 +604,13 @@ public class DateTimeFunctions {
* "-P6H3M" -- parses as "-6 hours and -3 minutes"
* "-P-6H+3M" -- parses as "+6 hours and -3 minutes"
*/
- @ScalarFunction
+ @ScalarFunction(volatility = FunctionVolatility.VOLATILE)
public static long ago(String periodString) {
Duration period = Duration.parse(periodString);
return System.currentTimeMillis() - period.toMillis();
}
- @ScalarFunction
+ @ScalarFunction(volatility = FunctionVolatility.VOLATILE)
public static long[] agoMV(String[] periodString) {
long[] results = new long[periodString.length];
for (int i = 0; i < periodString.length; i++) {
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/InternalFunctions.java
b/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/InternalFunctions.java
index c2b2c0e4bee..235c861b990 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/InternalFunctions.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/InternalFunctions.java
@@ -18,10 +18,12 @@
*/
package org.apache.pinot.common.function.scalar;
+import org.apache.pinot.spi.annotations.FunctionVolatility;
import org.apache.pinot.spi.annotations.ScalarFunction;
import org.apache.pinot.spi.query.QueryThreadContext;
+@ScalarFunction(enabled = false, volatility = FunctionVolatility.STABLE)
public class InternalFunctions {
private InternalFunctions() {
}
@@ -33,7 +35,7 @@ public class InternalFunctions {
/// input.
///
/// This is mostly useful for test and internal usage
- @ScalarFunction
+ @ScalarFunction(volatility = FunctionVolatility.VOLATILE)
public static String cid(String input) {
return QueryThreadContext.get().getExecutionContext().getCid();
}
@@ -57,7 +59,7 @@ public class InternalFunctions {
/// input.
///
/// This is mostly useful for test and internal usage and should be close to
now()
- @ScalarFunction
+ @ScalarFunction(volatility = FunctionVolatility.VOLATILE)
public static long startTime(String input) {
return QueryThreadContext.get().getExecutionContext().getStartTimeMs();
}
@@ -69,7 +71,7 @@ public class InternalFunctions {
/// input.
///
/// This is mostly useful for test and internal usage
- @ScalarFunction
+ @ScalarFunction(volatility = FunctionVolatility.VOLATILE)
public static long endTime(String input) {
return
QueryThreadContext.get().getExecutionContext().getActiveDeadlineMs();
}
@@ -102,7 +104,7 @@ public class InternalFunctions {
/// input.
///
/// This is mostly useful for test and internal usage
- @ScalarFunction
+ @ScalarFunction(volatility = FunctionVolatility.VOLATILE)
public static int stageId(String input) {
QueryThreadContext.MseWorkerInfo mseWorkerInfo =
QueryThreadContext.get().getMseWorkerInfo();
return mseWorkerInfo != null ? mseWorkerInfo.getStageId() : -1;
@@ -116,7 +118,7 @@ public class InternalFunctions {
/// input.
///
/// This is mostly useful for test and internal usage
- @ScalarFunction
+ @ScalarFunction(volatility = FunctionVolatility.VOLATILE)
public static int workerId(String input) {
QueryThreadContext.MseWorkerInfo mseWorkerInfo =
QueryThreadContext.get().getMseWorkerInfo();
return mseWorkerInfo != null ? mseWorkerInfo.getWorkerId() : -1;
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/function/FunctionUtilsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/function/FunctionUtilsTest.java
index ef87afa6687..3d1b6a61e3c 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/function/FunctionUtilsTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/function/FunctionUtilsTest.java
@@ -26,14 +26,44 @@ import java.util.HashMap;
import java.util.List;
import java.util.UUID;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.spi.annotations.FunctionVolatility;
+import org.apache.pinot.spi.annotations.ScalarFunction;
import org.apache.pinot.spi.utils.PinotDataType;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
public class FunctionUtilsTest {
+ /**
+ * Test fixture for class-level function volatility.
+ */
+ @ScalarFunction(enabled = false, volatility = FunctionVolatility.STABLE)
+ private static class AnnotatedFunction {
+ @ScalarFunction(enabled = false)
+ public static long classAnnotated() {
+ return 0L;
+ }
+ }
+
+ /**
+ * Test fixture for method-level and legacy volatility metadata.
+ */
+ private static class VolatilityAnnotatedFunction {
+ @ScalarFunction(enabled = false, isDeterministic = false)
+ public static long legacyVolatile() {
+ return 0L;
+ }
+
+ @ScalarFunction(enabled = false, volatility = FunctionVolatility.STABLE)
+ public static long stable() {
+ return 0L;
+ }
+ }
@Test
public void testGetArgumentType() {
@@ -56,6 +86,100 @@ public class FunctionUtilsTest {
assertEquals(FunctionUtils.getArgumentType((short) 1),
PinotDataType.SHORT);
}
+ @Test
+ public void testFunctionVolatilityMetadata() {
+ FunctionInfo now = FunctionRegistry.lookupFunctionInfo("now", 0);
+ assertTrue(now.isDeterministic());
+ assertEquals(now.getVolatility(), FunctionVolatility.VOLATILE);
+
+ FunctionInfo ago = FunctionRegistry.lookupFunctionInfo("ago", 1);
+ assertTrue(ago.isDeterministic());
+ assertEquals(ago.getVolatility(), FunctionVolatility.VOLATILE);
+
+ FunctionInfo agoMv = FunctionRegistry.lookupFunctionInfo("agomv", 1);
+ assertTrue(agoMv.isDeterministic());
+ assertEquals(agoMv.getVolatility(), FunctionVolatility.VOLATILE);
+
+ FunctionInfo random = FunctionRegistry.lookupFunctionInfo("rand", 0);
+ assertFalse(random.isDeterministic());
+ assertEquals(random.getVolatility(), FunctionVolatility.VOLATILE);
+
+ FunctionInfo seededRandom = FunctionRegistry.lookupFunctionInfo("rand", 1);
+ assertTrue(seededRandom.isDeterministic());
+ assertEquals(seededRandom.getVolatility(), FunctionVolatility.IMMUTABLE);
+
+ FunctionInfo sleep = FunctionRegistry.lookupFunctionInfo("sleep", 1);
+ assertTrue(sleep.isDeterministic());
+ assertEquals(sleep.getVolatility(), FunctionVolatility.VOLATILE);
+
+ FunctionInfo cid = FunctionRegistry.lookupFunctionInfo("cid", 1);
+ assertTrue(cid.isDeterministic());
+ assertEquals(cid.getVolatility(), FunctionVolatility.VOLATILE);
+
+ FunctionInfo requestId = FunctionRegistry.lookupFunctionInfo("reqid", 1);
+ assertTrue(requestId.isDeterministic());
+ assertEquals(requestId.getVolatility(), FunctionVolatility.STABLE);
+
+ FunctionInfo startTime = FunctionRegistry.lookupFunctionInfo("starttime",
1);
+ assertTrue(startTime.isDeterministic());
+ assertEquals(startTime.getVolatility(), FunctionVolatility.VOLATILE);
+
+ FunctionInfo workerId = FunctionRegistry.lookupFunctionInfo("workerid", 1);
+ assertTrue(workerId.isDeterministic());
+ assertEquals(workerId.getVolatility(), FunctionVolatility.VOLATILE);
+ }
+
+ @Test
+ public void testFunctionVolatilityResolution()
+ throws NoSuchMethodException {
+ FunctionInfo classAnnotated =
+
FunctionInfo.fromMethod(AnnotatedFunction.class.getMethod("classAnnotated"));
+ assertTrue(classAnnotated.isDeterministic());
+ assertEquals(classAnnotated.getVolatility(), FunctionVolatility.STABLE);
+
+ FunctionInfo classAnnotationAwareConstructor =
+ new FunctionInfo(AnnotatedFunction.class.getMethod("classAnnotated"),
AnnotatedFunction.class, false);
+ assertTrue(classAnnotationAwareConstructor.isDeterministic());
+ assertEquals(classAnnotationAwareConstructor.getVolatility(),
FunctionVolatility.STABLE);
+
+ FunctionInfo legacyVolatile =
+
FunctionInfo.fromMethod(VolatilityAnnotatedFunction.class.getMethod("legacyVolatile"));
+ assertFalse(legacyVolatile.isDeterministic());
+ assertEquals(legacyVolatile.getVolatility(), FunctionVolatility.VOLATILE);
+
+ FunctionInfo annotationAwareConstructor =
+ new
FunctionInfo(VolatilityAnnotatedFunction.class.getMethod("legacyVolatile"),
+ VolatilityAnnotatedFunction.class, false, true);
+ assertTrue(annotationAwareConstructor.isDeterministic());
+ assertEquals(annotationAwareConstructor.getVolatility(),
FunctionVolatility.VOLATILE);
+
+ FunctionInfo deterministicOverride =
+ new FunctionInfo(VolatilityAnnotatedFunction.class.getMethod("stable"),
+ VolatilityAnnotatedFunction.class, false, false);
+ assertFalse(deterministicOverride.isDeterministic());
+ assertEquals(deterministicOverride.getVolatility(),
FunctionVolatility.VOLATILE);
+
+ FunctionInfo explicitVolatilityOverride = new FunctionInfo(
+ AnnotatedFunction.class.getMethod("classAnnotated"),
AnnotatedFunction.class, false, true,
+ FunctionVolatility.IMMUTABLE);
+ assertTrue(explicitVolatilityOverride.isDeterministic());
+ assertEquals(explicitVolatilityOverride.getVolatility(),
FunctionVolatility.IMMUTABLE);
+
+ PinotScalarFunction scalarFunction =
+
PinotScalarFunction.fromMethod(VolatilityAnnotatedFunction.class.getMethod("stable"),
false, true);
+ FunctionInfo dynamicFunctionInfo = scalarFunction.getFunctionInfo(0);
+ assertNotNull(dynamicFunctionInfo);
+ assertTrue(dynamicFunctionInfo.isDeterministic());
+ assertEquals(dynamicFunctionInfo.getVolatility(),
FunctionVolatility.STABLE);
+
+ PinotScalarFunction legacyDynamicFunction =
+
PinotScalarFunction.fromMethod(VolatilityAnnotatedFunction.class.getMethod("legacyVolatile"),
false, true);
+ FunctionInfo legacyDynamicFunctionInfo =
legacyDynamicFunction.getFunctionInfo(0);
+ assertNotNull(legacyDynamicFunctionInfo);
+ assertTrue(legacyDynamicFunctionInfo.isDeterministic());
+ assertEquals(legacyDynamicFunctionInfo.getVolatility(),
FunctionVolatility.VOLATILE);
+ }
+
@Test
public void testGetArgumentTypeForVendorTimestampSubclass() {
// Vendor JDBC drivers commonly return Timestamp subclasses (e.g. BigQuery
Simba's TimestampTz).
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSchemaRestletResource.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSchemaRestletResource.java
index 7cf819a9cd6..f7bc6d349ef 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSchemaRestletResource.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSchemaRestletResource.java
@@ -409,7 +409,8 @@ public class PinotSchemaRestletResource {
try {
List<TableConfig> tableConfigs =
_pinotHelixResourceManager.getTableConfigsForSchema(schema.getSchemaName());
boolean isIgnoreCase =
_pinotHelixResourceManager.getTableCache().isIgnoreCase();
- SchemaUtils.validate(schema, tableConfigs, isIgnoreCase);
+ Schema existingSchema =
_pinotHelixResourceManager.getSchema(schema.getSchemaName());
+ SchemaUtils.validate(schema, tableConfigs, isIgnoreCase, existingSchema);
} catch (Exception e) {
throw new ControllerApplicationException(LOGGER,
"Invalid schema: " + schema.getSchemaName() + ". Reason: " +
e.getMessage(), Response.Status.BAD_REQUEST, e);
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java
index 035c36a1243..550fc3e132f 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java
@@ -339,15 +339,24 @@ public class PinotTableRestletResource {
boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
boolean hasRealtime = tableConfigNode.has(TableType.REALTIME.name());
- if (hasOffline && !hasRealtime) {
- throw new IllegalStateException("pure offline table copy not supported
yet");
- }
+ TableConfig realtimeTableConfig;
+ try {
+ if (hasOffline && !hasRealtime) {
+ throw new IllegalStateException("pure offline table copy not
supported yet");
+ }
- ObjectNode realtimeTableConfigNode = (ObjectNode)
tableConfigNode.get(TableType.REALTIME.name());
- tweakRealtimeTableConfig(realtimeTableConfigNode, copyTablePayload);
- TableConfig realtimeTableConfig =
JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
- if (realtimeTableConfig.getUpsertConfig() != null) {
- throw new IllegalStateException("upsert table copy not supported");
+ ObjectNode realtimeTableConfigNode = (ObjectNode)
tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, copyTablePayload);
+ realtimeTableConfig =
JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ if (realtimeTableConfig.getUpsertConfig() != null) {
+ throw new IllegalStateException("upsert table copy not supported");
+ }
+ // Run the complete validation stack before dry-run returns and before
the schema is persisted. This also
+ // prevents copying a legacy non-immutable transform into a new
ingestion pipeline.
+ TableConfigValidationUtils.validateTableConfig(
+ realtimeTableConfig, schema, null, _pinotHelixResourceManager,
_controllerConf, _pinotTaskManager);
+ } catch (ConfigValidationException | IllegalArgumentException |
IllegalStateException e) {
+ throw new ControllerApplicationException(LOGGER, e.getMessage(),
Response.Status.BAD_REQUEST, e);
}
LOGGER.info("[copyTable] Successfully fetched and tweaked table config
for table: {}", tableName);
@@ -360,8 +369,6 @@ public class PinotTableRestletResource {
_pinotHelixResourceManager.addSchema(schema, true, false);
LOGGER.info("[copyTable] Successfully added schema for table: {}",
tableName);
- TableConfigValidationUtils.validateTableConfig(
- realtimeTableConfig, schema, null, _pinotHelixResourceManager,
_controllerConf, _pinotTaskManager);
// Add the table with designated starting kafka offset and segment
sequence number to create consuming segments
_pinotHelixResourceManager.addTable(realtimeTableConfig,
streamMetadataList);
LOGGER.info("[copyTable] Successfully added table config: {} with
designated high watermark", tableName);
@@ -376,8 +383,8 @@ public class PinotTableRestletResource {
response.setWatermarkInductionResult(watermarkInductionResult);
}
return response;
- } catch (ConfigValidationException e) {
- throw new ControllerApplicationException(LOGGER, e.getMessage(),
Response.Status.BAD_REQUEST, e);
+ } catch (ControllerApplicationException e) {
+ throw e;
} catch (Exception e) {
LOGGER.error("[copyTable] Error copying table: {}", tableName, e);
throw new ControllerApplicationException(LOGGER, "Error copying table: "
+ e.getMessage(),
@@ -795,7 +802,8 @@ public class PinotTableRestletResource {
schema = _pinotHelixResourceManager.getTableSchema(tableNameWithType);
Preconditions.checkState(schema != null, "Failed to find schema for
table: %s", tableNameWithType);
TableConfigValidationUtils.validateTableConfig(
- tableConfig, schema, typesToSkip, _pinotHelixResourceManager,
_controllerConf, _pinotTaskManager);
+ tableConfig, schema, typesToSkip, _pinotHelixResourceManager,
_controllerConf, _pinotTaskManager,
+ _pinotHelixResourceManager.getTableConfig(tableNameWithType));
} catch (Exception e) {
String msg = String.format("Invalid table config: %s with error: %s",
tableName, e.getMessage());
throw new ControllerApplicationException(LOGGER, msg,
Response.Status.BAD_REQUEST, e);
@@ -864,7 +872,8 @@ public class PinotTableRestletResource {
if (schema == null) {
throw new SchemaNotFoundException("Failed to find schema for table: "
+ tableNameWithType);
}
- TableConfigUtils.validate(tableConfig, schema, typesToSkip);
+ TableConfigUtils.validate(tableConfig, schema, typesToSkip,
+ _pinotHelixResourceManager.getTableConfig(tableNameWithType));
TaskConfigUtils.validateTaskConfigs(tableConfig, schema,
_pinotTaskManager, typesToSkip);
ObjectNode tableConfigValidateStr = JsonUtils.newObjectNode();
if (tableConfig.getTableType() == TableType.OFFLINE) {
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtils.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtils.java
index ec51fbccec7..f3d96d3c741 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtils.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtils.java
@@ -59,8 +59,15 @@ public final class TableConfigValidationUtils {
public static void validateTableConfig(TableConfig tableConfig, Schema
schema,
@Nullable String typesToSkip, PinotHelixResourceManager resourceManager,
ControllerConf controllerConf, @Nullable PinotTaskManager taskManager) {
+ validateTableConfig(tableConfig, schema, typesToSkip, resourceManager,
controllerConf, taskManager, null);
+ }
+
+ public static void validateTableConfig(TableConfig tableConfig, Schema
schema,
+ @Nullable String typesToSkip, PinotHelixResourceManager resourceManager,
+ ControllerConf controllerConf, @Nullable PinotTaskManager taskManager,
+ @Nullable TableConfig existingTableConfig) {
validateEnvironmentVariables(tableConfig);
- TableConfigUtils.validate(tableConfig, schema, typesToSkip);
+ TableConfigUtils.validate(tableConfig, schema, typesToSkip,
existingTableConfig);
TableConfigUtils.validateTableName(tableConfig);
TableConfigUtils.ensureMinReplicas(tableConfig,
controllerConf.getDefaultTableMinReplicas());
TableConfigUtils.ensureStorageQuotaConstraints(tableConfig,
controllerConf.getDimTableMaxSize());
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigsRestletResource.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigsRestletResource.java
index 17e7dab0471..a74d52fc9ad 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigsRestletResource.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigsRestletResource.java
@@ -583,6 +583,7 @@ public class TableConfigsRestletResource {
Preconditions.checkState(rawTableName.equals(schemaName),
"'tableName': %s must be equal to 'schemaName' from 'schema': %s",
rawTableName, schema.getSchemaName());
+ SchemaUtils.validateIngestionTransformVolatility(schema,
_pinotHelixResourceManager.getSchema(schemaName));
SchemaUtils.validate(schema);
if (offlineTableConfig != null) {
String offlineRawTableName = DatabaseUtils.translateTableName(
@@ -590,7 +591,8 @@ public class TableConfigsRestletResource {
Preconditions.checkState(offlineRawTableName.equals(rawTableName),
"Name in 'offline' table config: %s must be equal to 'tableName':
%s", offlineRawTableName, rawTableName);
TableConfigUtils.validateTableName(offlineTableConfig);
- TableConfigUtils.validate(offlineTableConfig, schema, typesToSkip);
+ TableConfigUtils.validate(offlineTableConfig, schema, typesToSkip,
+
_pinotHelixResourceManager.getTableConfig(TableNameBuilder.OFFLINE.tableNameWithType(rawTableName)));
TaskConfigUtils.validateTaskConfigs(tableConfigs.getOffline(), schema,
_pinotTaskManager, typesToSkip);
TableConfigValidatorRegistry.validate(offlineTableConfig, schema);
}
@@ -600,7 +602,8 @@ public class TableConfigsRestletResource {
Preconditions.checkState(realtimeRawTableName.equals(rawTableName),
"Name in 'realtime' table config: %s must be equal to 'tableName':
%s", realtimeRawTableName, rawTableName);
TableConfigUtils.validateTableName(realtimeTableConfig);
- TableConfigUtils.validate(realtimeTableConfig, schema, typesToSkip);
+ TableConfigUtils.validate(realtimeTableConfig, schema, typesToSkip,
+
_pinotHelixResourceManager.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(rawTableName)));
TaskConfigUtils.validateTaskConfigs(tableConfigs.getRealtime(),
schema, _pinotTaskManager, typesToSkip);
TableConfigValidatorRegistry.validate(realtimeTableConfig, schema);
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotSchemaRestletResourceTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotSchemaRestletResourceTest.java
index 2dc7f7356c6..0e41edc7cbd 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotSchemaRestletResourceTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotSchemaRestletResourceTest.java
@@ -260,6 +260,47 @@ public class PinotSchemaRestletResourceTest {
"Expected TimeFieldSpec error, got: " + validateCause.getMessage());
}
+ @Test
+ public void testNonDeterministicSchemaTransformCreateAndLegacyUpdate()
+ throws Exception {
+ PinotAdminClient adminClient = TEST_INSTANCE.getOrCreateAdminClient();
+ String schemaName = "legacyNonDeterministicSchemaTransform";
+ Schema schema = TEST_INSTANCE.createDummySchema(schemaName);
+ DimensionFieldSpec eventTimeField = new DimensionFieldSpec("eventTimeMs",
DataType.LONG, true);
+ eventTimeField.setTransformFunction("now()");
+ schema.addField(eventTimeField);
+ try {
+ expectValidationExceptionWithMessage(
+ () ->
adminClient.getSchemaClient().createSchema(schema.toSingleLineJsonString()),
+ "Function 'now' has VOLATILE volatility");
+ expectValidationExceptionWithMessage(
+ () ->
adminClient.getSchemaClient().validateSchema(schema.toSingleLineJsonString()),
+ "Function 'now' has VOLATILE volatility");
+
+ // Seed below the REST validation layer to model a schema persisted
before this validation existed.
+ TEST_INSTANCE.getHelixResourceManager().addSchema(schema, false, false);
+
+ Schema update =
adminClient.getSchemaClient().getSchemaObject(schemaName);
+ update.addField(new DimensionFieldSpec("newColumn", DataType.STRING,
true));
+
adminClient.getSchemaClient().validateSchema(update.toSingleLineJsonString());
+ adminClient.getSchemaClient().updateSchema(schemaName,
update.toSingleLineJsonString());
+
+ Schema stored =
adminClient.getSchemaClient().getSchemaObject(schemaName);
+ assertTrue(stored.hasColumn("newColumn"));
+
assertEquals(stored.getFieldSpecFor("eventTimeMs").getTransformFunction(),
"now()");
+
+ update.getFieldSpecFor("eventTimeMs").setTransformFunction("plus(now(),
1)");
+ expectValidationExceptionWithMessage(
+ () ->
adminClient.getSchemaClient().validateSchema(update.toSingleLineJsonString()),
+ "Function 'now' has VOLATILE volatility");
+ expectValidationExceptionWithMessage(
+ () -> adminClient.getSchemaClient().updateSchema(schemaName,
update.toSingleLineJsonString()),
+ "Function 'now' has VOLATILE volatility");
+ } finally {
+ TEST_INSTANCE.getHelixResourceManager().deleteSchema(schemaName);
+ }
+ }
+
@Test
public void testSchemaDeletionWithLogicalTable()
throws Exception {
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTableRestletResourceTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTableRestletResourceTest.java
index cb29afe3d20..33b62aa4621 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTableRestletResourceTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTableRestletResourceTest.java
@@ -58,6 +58,8 @@ import
org.apache.pinot.spi.config.table.assignment.InstanceConstraintConfig;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import
org.apache.pinot.spi.config.table.assignment.InstanceReplicaGroupPartitionConfig;
import org.apache.pinot.spi.config.table.assignment.InstanceTagPoolConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.LogicalTableConfig;
import org.apache.pinot.spi.data.Schema;
@@ -530,6 +532,67 @@ public class PinotTableRestletResourceTest extends
ControllerTest {
}
}
+ @Test
+ public void testNonDeterministicTransformCreateAndLegacyUpdate()
+ throws Exception {
+ String tableName = "legacyNonDeterministicTransform";
+ DEFAULT_INSTANCE.addDummySchema(tableName);
+
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setTransformConfigs(List.of(new TransformConfig("dimA",
"now()")));
+ TableConfig legacyTableConfig = getOfflineTableBuilder(tableName)
+ .setIngestionConfig(ingestionConfig)
+ .build();
+
+ IOException createError =
+ expectThrows(IOException.class, () ->
createTable(legacyTableConfig.toJsonString()));
+ assertHasStatus(createError, 400);
+ assertTrue(createError.getMessage().contains("Function 'now' has VOLATILE
volatility"), createError.getMessage());
+
+ // Seed the config below the REST validation layer to model a table
persisted before this validation existed.
+ DEFAULT_INSTANCE.getHelixResourceManager().addTable(legacyTableConfig);
+
+ TableConfig update = getTableConfig(tableName, "OFFLINE");
+ update.getValidationConfig().setRetentionTimeValue("10");
+ JsonNode validationResponse =
JsonUtils.stringToJsonNode(tableClient().validateTableConfig(update.toJsonString()));
+ assertTrue(validationResponse.has("OFFLINE"));
+
+ JsonNode response = JsonUtils.stringToJsonNode(updateTable(tableName,
update.toJsonString()));
+ assertTrue(response.has("status"));
+
+ TableConfig stored = getTableConfig(tableName, "OFFLINE");
+ assertEquals(stored.getValidationConfig().getRetentionTimeValue(), "10");
+
assertEquals(stored.getIngestionConfig().getTransformConfigs().get(0).getTransformFunction(),
"now()");
+
+ IngestionConfig changedIngestionConfig = new IngestionConfig();
+ changedIngestionConfig.setTransformConfigs(List.of(new
TransformConfig("dimA", "plus(now(), 1)")));
+ update.setIngestionConfig(changedIngestionConfig);
+ PinotAdminException validationError =
+ expectThrows(PinotAdminException.class, () ->
tableClient().validateTableConfig(update.toJsonString()));
+ assertTrue(validationError.getMessage().contains("Function 'now' has
VOLATILE volatility"),
+ validationError.getMessage());
+
+ IOException updateError = expectThrows(IOException.class, () ->
updateTable(tableName, update.toJsonString()));
+ assertHasStatus(updateError, 400);
+ assertTrue(updateError.getMessage().contains("Function 'now' has VOLATILE
volatility"), updateError.getMessage());
+ }
+
+ @Test
+ public void testCreateTableRejectsNonDeterministicSchemaTransform()
+ throws Exception {
+ String tableName = "nonDeterministicSchemaTransform";
+ Schema schema = DEFAULT_INSTANCE.createDummySchema(tableName);
+ schema.getFieldSpecFor("dimA").setTransformFunction("now()");
+
+ // Seed below the schema REST validation layer to model a schema persisted
before this validation existed.
+ DEFAULT_INSTANCE.getHelixResourceManager().addSchema(schema, false, false);
+
+ TableConfig tableConfig = getOfflineTableBuilder(tableName).build();
+ IOException createError = expectThrows(IOException.class, () ->
createTable(tableConfig.toJsonString()));
+ assertHasStatus(createError, 400);
+ assertTrue(createError.getMessage().contains("Function 'now' has VOLATILE
volatility"), createError.getMessage());
+ }
+
private void deleteAllTables()
throws IOException {
List<String> tables = getTableNames("offline", null, null, null);
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
index b7d6268ba56..7537fa65e66 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
@@ -33,7 +33,10 @@ import
org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.TunerConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DimensionFieldSpec;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.LogicalTableConfig;
import org.apache.pinot.spi.data.MetricFieldSpec;
@@ -512,6 +515,106 @@ public class TableConfigsRestletResourceTest extends
ControllerTest {
adminClient.getTableClient().deleteTableConfigs(tableName, null);
}
+ @Test
+ public void testNonDeterministicTransformCreateAndLegacyUpdate()
+ throws Exception {
+ PinotAdminClient adminClient = getOrCreateAdminClient();
+ String tableName = "legacyNonDeterministicTableConfigs";
+ Schema schema = createDummySchema(tableName);
+
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setTransformConfigs(List.of(new TransformConfig("dimA",
"now()")));
+ TableConfig legacyOfflineConfig = getBaseTableConfigBuilder(tableName,
TableType.OFFLINE)
+ .setIngestionConfig(ingestionConfig)
+ .build();
+ TableConfigs tableConfigs = new TableConfigs(tableName, schema,
legacyOfflineConfig, null);
+ String tableNameWithType =
TableNameBuilder.OFFLINE.tableNameWithType(tableName);
+ try {
+ String createError = Assert.expectThrows(Exception.class,
+ () ->
adminClient.getTableClient().createTableConfigs(tableConfigs.toPrettyJsonString(),
null, null))
+ .getMessage();
+ Assert.assertTrue(createError.contains("Function 'now' has VOLATILE
volatility"), createError);
+
+ // Seed the config below the REST validation layer to model a table
persisted before this validation existed.
+ DEFAULT_INSTANCE.addSchema(schema);
+ DEFAULT_INSTANCE.getHelixResourceManager().addTable(legacyOfflineConfig);
+
+ TableConfigs update =
adminClient.getTableClient().getTableConfigsObject(tableName);
+ update.getOffline().getValidationConfig().setRetentionTimeValue("10");
+ adminClient.getTableClient()
+ .updateTableConfigs(tableName, update.toPrettyJsonString(), null,
false, false);
+
+ TableConfigs stored =
adminClient.getTableClient().getTableConfigsObject(tableName);
+
Assert.assertEquals(stored.getOffline().getValidationConfig().getRetentionTimeValue(),
"10");
+
Assert.assertEquals(stored.getOffline().getIngestionConfig().getTransformConfigs().get(0).getTransformFunction(),
+ "now()");
+
+ IngestionConfig changedIngestionConfig = new IngestionConfig();
+ changedIngestionConfig.setTransformConfigs(List.of(new
TransformConfig("dimA", "plus(now(), 1)")));
+ update.getOffline().setIngestionConfig(changedIngestionConfig);
+ String updateError = Assert.expectThrows(Exception.class,
+ () -> adminClient.getTableClient()
+ .updateTableConfigs(tableName, update.toPrettyJsonString(),
null, false, false))
+ .getMessage();
+ Assert.assertTrue(updateError.contains("Function 'now' has VOLATILE
volatility"), updateError);
+ } finally {
+ if
(DEFAULT_INSTANCE.getHelixResourceManager().hasTable(tableNameWithType)) {
+ adminClient.getTableClient().deleteTableConfigs(tableName, null);
+ } else if
(DEFAULT_INSTANCE.getHelixResourceManager().getSchema(tableName) != null) {
+ adminClient.getSchemaClient().deleteSchema(tableName);
+ }
+ }
+ }
+
+ @Test
+ public void testNonDeterministicSchemaTransformCreateAndLegacyUpdate()
+ throws Exception {
+ PinotAdminClient adminClient = getOrCreateAdminClient();
+ String tableName = "legacyNonDeterministicSchemaTableConfigs";
+ Schema schema = createDummySchema(tableName);
+ schema.getFieldSpecFor("dimA").setTransformFunction("now()");
+ TableConfig legacyOfflineConfig = getBaseTableConfigBuilder(tableName,
TableType.OFFLINE).build();
+ TableConfigs tableConfigs = new TableConfigs(tableName, schema,
legacyOfflineConfig, null);
+ String tableNameWithType =
TableNameBuilder.OFFLINE.tableNameWithType(tableName);
+ try {
+ String createError = Assert.expectThrows(Exception.class,
+ () ->
adminClient.getTableClient().createTableConfigs(tableConfigs.toPrettyJsonString(),
null, null))
+ .getMessage();
+ Assert.assertTrue(createError.contains("Function 'now' has VOLATILE
volatility"), createError);
+
+ // Seed below the REST validation layer to model a table and schema
persisted before this validation existed.
+ DEFAULT_INSTANCE.getHelixResourceManager().addSchema(schema, false,
false);
+ DEFAULT_INSTANCE.getHelixResourceManager().addTable(legacyOfflineConfig);
+
+ TableConfigs update =
adminClient.getTableClient().getTableConfigsObject(tableName);
+ update.getSchema().addField(new DimensionFieldSpec("newColumn",
FieldSpec.DataType.STRING, true));
+
adminClient.getTableClient().validateTableConfigs(update.toPrettyJsonString(),
null);
+ adminClient.getTableClient()
+ .updateTableConfigs(tableName, update.toPrettyJsonString(), null,
false, false);
+
+ TableConfigs stored =
adminClient.getTableClient().getTableConfigsObject(tableName);
+ Assert.assertTrue(stored.getSchema().hasColumn("newColumn"));
+
Assert.assertEquals(stored.getSchema().getFieldSpecFor("dimA").getTransformFunction(),
"now()");
+
+
update.getSchema().getFieldSpecFor("dimA").setTransformFunction("plus(now(),
1)");
+ String validationError = Assert.expectThrows(Exception.class,
+ () ->
adminClient.getTableClient().validateTableConfigs(update.toPrettyJsonString(),
null))
+ .getMessage();
+ Assert.assertTrue(validationError.contains("Function 'now' has VOLATILE
volatility"), validationError);
+ String updateError = Assert.expectThrows(Exception.class,
+ () -> adminClient.getTableClient()
+ .updateTableConfigs(tableName, update.toPrettyJsonString(),
null, false, false))
+ .getMessage();
+ Assert.assertTrue(updateError.contains("Function 'now' has VOLATILE
volatility"), updateError);
+ } finally {
+ if
(DEFAULT_INSTANCE.getHelixResourceManager().hasTable(tableNameWithType)) {
+ adminClient.getTableClient().deleteTableConfigs(tableName, null);
+ } else if
(DEFAULT_INSTANCE.getHelixResourceManager().getSchema(tableName) != null) {
+ adminClient.getSchemaClient().deleteSchema(tableName);
+ }
+ }
+ }
+
@Test
public void testForceUpdateTableSchemaAndConfigs()
throws Exception {
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/util/SchemaUtilsTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/util/SchemaUtilsTest.java
index f2d4f07df81..c17e1b69b6e 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/util/SchemaUtilsTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/util/SchemaUtilsTest.java
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.pinot.segment.local.utils.SchemaUtils;
+import org.apache.pinot.segment.local.utils.TableConfigUtils;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
@@ -189,6 +190,53 @@ public class SchemaUtilsTest {
SchemaUtils.validate(schema, Lists.newArrayList(tableConfig));
}
+ @Test
+ public void testCompatibilityGrandfathersExistingNonDeterministicTransform()
{
+ Schema schema =
+ new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addMetric("eventTimeMs",
DataType.LONG).build();
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setTransformConfigs(List.of(new
TransformConfig("eventTimeMs", "now()")));
+ TableConfig tableConfig =
+ new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(ingestionConfig).build();
+
+ // A new table using this transform is still rejected.
+ Assert.expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validate(tableConfig, schema));
+
+ // Schema validation checks an already-associated, unchanged table config,
so the legacy transform is
+ // grandfathered while the rest of table/schema compatibility validation
continues to run.
+ SchemaUtils.validate(schema, List.of(tableConfig));
+ }
+
+ @Test
+ public void testSchemaTransformVolatility()
+ throws Exception {
+ Schema schema =
+ new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addMetric("eventTimeMs",
DataType.LONG).build();
+ schema.getFieldSpecFor("eventTimeMs").setTransformFunction("now()");
+
+ // Runtime validation remains permissive so a legacy schema already stored
in ZK can still be loaded.
+ SchemaUtils.validate(schema);
+
+ IllegalStateException nonDeterministicError =
Assert.expectThrows(IllegalStateException.class,
+ () -> SchemaUtils.validate(schema, List.of()));
+ Assert.assertTrue(nonDeterministicError.getMessage().contains("Function
'now' has VOLATILE volatility"),
+ nonDeterministicError.getMessage());
+
+ Schema existingSchema = Schema.fromString(schema.toString());
+ Schema unchangedTransform = Schema.fromString(schema.toString());
+ unchangedTransform.addField(new DimensionFieldSpec("newColumn",
DataType.STRING, true));
+ SchemaUtils.validate(unchangedTransform, List.of(), false, existingSchema);
+
+
unchangedTransform.getFieldSpecFor("eventTimeMs").setTransformFunction("plus(now(),
1)");
+ nonDeterministicError = Assert.expectThrows(IllegalStateException.class,
+ () -> SchemaUtils.validate(unchangedTransform, List.of(), false,
existingSchema));
+ Assert.assertTrue(nonDeterministicError.getMessage().contains("Function
'now' has VOLATILE volatility"),
+ nonDeterministicError.getMessage());
+
+
unchangedTransform.getFieldSpecFor("eventTimeMs").setTransformFunction("rand(123)");
+ SchemaUtils.validate(unchangedTransform, List.of(), false, existingSchema);
+ }
+
private Map<String, String> getStreamConfigs() {
Map<String, String> streamConfigs = new HashMap<>();
streamConfigs.put("streamType", "kafka");
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CLPEncodingRealtimeTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CLPEncodingRealtimeTest.java
index 68556333a24..356864cddc4 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CLPEncodingRealtimeTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CLPEncodingRealtimeTest.java
@@ -116,7 +116,7 @@ public class CLPEncodingRealtimeTest extends
CustomDataQueryClusterIntegrationTe
@Override
protected IngestionConfig getIngestionConfig() {
List<TransformConfig> transforms = new ArrayList<>();
- transforms.add(new TransformConfig("timestampInEpoch", "now()"));
+ transforms.add(new TransformConfig("timestampInEpoch", "1704067200000"));
IngestionConfig ingestionConfig = new IngestionConfig();
ingestionConfig.setTransformConfigs(transforms);
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryPlannerRuleOptionsTest.java
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryPlannerRuleOptionsTest.java
index 004175b17de..b9de8291778 100644
---
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryPlannerRuleOptionsTest.java
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryPlannerRuleOptionsTest.java
@@ -222,10 +222,13 @@ public class QueryPlannerRuleOptionsTest extends
QueryEnvironmentTestBase {
}
@Test
- public void testRandFunctionNotEvaluatedInMultiStagePlanner() {
- String query = "EXPLAIN PLAN FOR SELECT rand() FROM b";
+ public void
testFunctionVolatilityDoesNotChangeLiteralEvaluationInMultiStagePlanner() {
+ String query = "EXPLAIN PLAN FOR SELECT now(), ago('PT1H'), rand(),
rand(123) FROM b";
String explain = _queryEnvironment.explainQuery(query,
RANDOM_REQUEST_ID_GEN.nextLong());
assertTrue(explain.contains("RAND()"), "Expected RAND() to remain in
logical plan");
+ assertFalse(explain.contains("NOW()"), "Expected NOW() to be evaluated
into a literal");
+ assertFalse(explain.contains("AGO("), "Expected AGO('PT1H') to be
evaluated into a literal");
+ assertFalse(explain.contains("RAND(123)"), "Expected RAND(123) to be
evaluated into a literal");
}
@Test
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SchemaUtils.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SchemaUtils.java
index 99ec4d9ddc2..2147d64c5d8 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SchemaUtils.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SchemaUtils.java
@@ -25,7 +25,9 @@ import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
+import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.common.evaluator.FunctionEvaluatorFactory;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -71,21 +73,55 @@ public class SchemaUtils {
* @param tableConfigs table configs associated with this schema (table
configs with raw name = schema name)
*/
public static void validate(Schema schema, List<TableConfig> tableConfigs) {
- validate(schema, tableConfigs, false);
+ validate(schema, tableConfigs, false, null);
}
public static void validate(Schema schema, List<TableConfig> tableConfigs,
boolean isIgnoreCase) {
+ validate(schema, tableConfigs, isIgnoreCase, null);
+ }
+
+ public static void validate(Schema schema, List<TableConfig> tableConfigs,
boolean isIgnoreCase,
+ @Nullable Schema existingSchema) {
// The deprecation reject is intentionally placed only in this REST-driven
overload (not in the single-arg
// `validate(Schema)` that server-side table loading uses), so existing
legacy schemas in ZK keep
// loading. New / updated schemas submitted via the controller REST API
are blocked.
// See https://github.com/apache/pinot/issues/2756 for the TimeFieldSpec
deprecation plan.
rejectDeprecatedTimeFieldSpec(schema);
+ validateIngestionTransformVolatility(schema, existingSchema);
for (TableConfig tableConfig : tableConfigs) {
validateCompatibilityWithTableConfig(schema, tableConfig);
}
validate(schema, isIgnoreCase);
}
+ /**
+ * Validates that new or changed legacy schema-level transform functions are
immutable. An exact transform on the
+ * same column in the stored schema is grandfathered so unrelated updates do
not strand existing schemas.
+ */
+ public static void validateIngestionTransformVolatility(Schema schema,
@Nullable Schema existingSchema) {
+ validateIngestionTransformVolatility(schema, existingSchema, Set.of(),
Set.of());
+ }
+
+ static void validateIngestionTransformVolatility(Schema schema, @Nullable
Schema existingSchema,
+ Set<String> overriddenColumns, Set<String> existingOverriddenColumns) {
+ for (FieldSpec fieldSpec : schema.getAllFieldSpecs()) {
+ if (fieldSpec.isVirtualColumn() ||
overriddenColumns.contains(fieldSpec.getName())) {
+ continue;
+ }
+ String transformFunction = fieldSpec.getTransformFunction();
+ if (transformFunction == null) {
+ continue;
+ }
+ FieldSpec existingFieldSpec =
+ existingSchema != null ?
existingSchema.getFieldSpecFor(fieldSpec.getName()) : null;
+ if (!existingOverriddenColumns.contains(fieldSpec.getName()) &&
existingFieldSpec != null
+ && Objects.equals(transformFunction,
existingFieldSpec.getTransformFunction())) {
+ continue;
+ }
+
TableConfigUtils.validateIngestionTransformFunctionVolatility(transformFunction);
+ }
+ }
+
@SuppressWarnings("deprecation")
private static void rejectDeprecatedTimeFieldSpec(Schema schema) {
if (schema.getTimeFieldSpec() != null) {
@@ -197,7 +233,10 @@ public class SchemaUtils {
*/
private static void validateCompatibilityWithTableConfig(Schema schema,
TableConfig tableConfig) {
try {
- TableConfigUtils.validate(tableConfig, schema);
+ // The associated table config already exists and is not being changed
by schema validation. Pass it as the
+ // existing config so unchanged legacy transforms remain grandfathered
while all schema-dependent validation
+ // still runs against the proposed schema.
+ TableConfigUtils.validate(tableConfig, schema, null, tableConfig);
} catch (Exception e) {
throw new IllegalStateException(
"Schema is incompatible with tableConfig with name: " +
tableConfig.getTableName() + " and type: "
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
index 31fbf48dc6c..403ec65deeb 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
@@ -39,6 +39,8 @@ import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.common.evaluator.FunctionEvaluatorFactory;
+import org.apache.pinot.common.function.FunctionInfo;
+import org.apache.pinot.common.function.FunctionRegistry;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.request.context.FunctionContext;
import org.apache.pinot.common.request.context.RequestContextUtils;
@@ -57,6 +59,7 @@ import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.StandardIndexes;
import
org.apache.pinot.segment.spi.index.multicolumntext.MultiColumnTextMetadata;
import
org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair;
+import org.apache.pinot.spi.annotations.FunctionVolatility;
import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
import org.apache.pinot.spi.config.table.DedupConfig;
import org.apache.pinot.spi.config.table.FieldConfig;
@@ -172,15 +175,20 @@ public final class TableConfigUtils {
* TODO: Add more validations for each section (e.g. validate conditions are
met for aggregateMetrics)
*/
public static void validate(TableConfig tableConfig, Schema schema,
@Nullable String typesToSkip) {
+ validate(tableConfig, schema, typesToSkip, null);
+ }
+
+ public static void validate(TableConfig tableConfig, Schema schema,
@Nullable String typesToSkip,
+ @Nullable TableConfig existingTableConfig) {
Preconditions.checkArgument(schema != null, "Schema should not be null for
table: %s", tableConfig.getTableName());
Set<ValidationType> skipTypes = parseTypesToSkipString(typesToSkip);
- validateEffectiveTableConfig(tableConfig, schema, skipTypes);
+ validateEffectiveTableConfig(tableConfig, schema, skipTypes,
existingTableConfig);
if (skipTypes.contains(ValidationType.ALL) ||
!hasConsumingSegmentTierOverwriteForRealtimeTable(tableConfig)) {
return;
}
try {
TableConfig consumingTableConfig =
overwriteTableConfigForConsumingSegmentTier(tableConfig);
- validateEffectiveTableConfig(consumingTableConfig, schema, skipTypes);
+ validateEffectiveTableConfig(consumingTableConfig, schema, skipTypes,
existingTableConfig);
} catch (RuntimeException e) {
throw new IllegalStateException(
"tierOverwrites.consuming produces an invalid table config: " +
e.getMessage(), e);
@@ -188,7 +196,7 @@ public final class TableConfigUtils {
}
private static void validateEffectiveTableConfig(TableConfig tableConfig,
Schema schema,
- Set<ValidationType> skipTypes) {
+ Set<ValidationType> skipTypes, @Nullable TableConfig
existingTableConfig) {
// Sanitize the table config before validation
sanitize(tableConfig);
@@ -196,9 +204,18 @@ public final class TableConfigUtils {
return;
}
+ // Schema-level transforms are active only when the table config does not
override the same destination. A
+ // non-immutable transform is grandfathered only when it was already
active for this table; removing an existing
+ // override must not silently activate it.
+ Set<String> overriddenColumns = getTransformColumns(tableConfig);
+ Set<String> existingOverriddenColumns =
getTransformColumns(existingTableConfig);
+ Schema existingSchema = existingTableConfig != null ? schema : null;
+ SchemaUtils.validateIngestionTransformVolatility(schema, existingSchema,
overriddenColumns,
+ existingOverriddenColumns);
+
validateValidationConfig(tableConfig, schema);
validateSegmentAssignmentConfig(tableConfig);
- validateIngestionConfig(tableConfig, schema);
+ validateIngestionConfig(tableConfig, schema, existingTableConfig);
if (tableConfig.getTableType() == TableType.REALTIME) {
validateStreamConfigMaps(tableConfig);
}
@@ -209,7 +226,7 @@ public final class TableConfigUtils {
validateInstanceAssignmentConfigs(tableConfig);
if (!skipTypes.contains(ValidationType.UPSERT)) {
- validateUpsertAndDedupConfig(tableConfig, schema);
+ validateUpsertAndDedupConfig(tableConfig, schema, existingTableConfig);
}
validateTaskConfig(tableConfig);
@@ -459,6 +476,12 @@ public final class TableConfigUtils {
/// - Schema-conforming transformer config.
@VisibleForTesting
public static void validateIngestionConfig(TableConfig tableConfig, Schema
schema) {
+ validateIngestionConfig(tableConfig, schema, null);
+ }
+
+ @VisibleForTesting
+ static void validateIngestionConfig(TableConfig tableConfig, Schema schema,
+ @Nullable TableConfig existingTableConfig) {
// All metrics-aggregation validation lives here; it returns the columns
referenced as aggregation sources, which
// a transform config is allowed to target as its destination (see the
transform validation below).
Set<String> aggregationSourceColumns =
validateMetricsAggregation(tableConfig, schema);
@@ -566,6 +589,7 @@ public final class TableConfigUtils {
// Transform configs
List<TransformConfig> transformConfigs =
ingestionConfig.getTransformConfigs();
if (transformConfigs != null) {
+ List<TransformConfig> existingTransformConfigs =
getTransformConfigs(existingTableConfig);
// Pre-pass: collect every column referenced as a transform-function
argument. A transform whose destination
// is not in the schema is still valid when another transform consumes
it as an input - i.e. it is an
// intermediate ("derived") column. This enables chained / parse-once
transforms, e.g.
@@ -611,6 +635,7 @@ public final class TableConfigUtils {
+ columnName + "'");
}
try {
+ validateIngestionTransformFunctionVolatility(transformConfig,
existingTransformConfigs);
expressionEvaluator =
FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction);
} catch (Exception e) {
throw new IllegalStateException(
@@ -649,6 +674,77 @@ public final class TableConfigUtils {
}
}
+ @Nullable
+ private static List<TransformConfig> getTransformConfigs(@Nullable
TableConfig tableConfig) {
+ IngestionConfig ingestionConfig = tableConfig != null ?
tableConfig.getIngestionConfig() : null;
+ return ingestionConfig != null ? ingestionConfig.getTransformConfigs() :
null;
+ }
+
+ private static Set<String> getTransformColumns(@Nullable TableConfig
tableConfig) {
+ List<TransformConfig> transformConfigs = getTransformConfigs(tableConfig);
+ return transformConfigs != null
+ ?
transformConfigs.stream().map(TransformConfig::getColumnName).filter(Objects::nonNull)
+ .collect(Collectors.toSet())
+ : Set.of();
+ }
+
+ @Nullable
+ private static List<TransformConfig>
getPostPartialUpsertTransformConfigs(@Nullable TableConfig tableConfig) {
+ UpsertConfig upsertConfig = tableConfig != null ?
tableConfig.getUpsertConfig() : null;
+ return upsertConfig != null ?
upsertConfig.getPostPartialUpsertTransformConfigs() : null;
+ }
+
+ private static void
validateIngestionTransformFunctionVolatility(TransformConfig transformConfig,
+ @Nullable List<TransformConfig> existingTransformConfigs) {
+ if (hasSameTransformConfig(existingTransformConfigs, transformConfig)) {
+ return;
+ }
+
validateIngestionTransformFunctionVolatility(transformConfig.getTransformFunction());
+ }
+
+ static void validateIngestionTransformFunctionVolatility(String
transformFunction) {
+ if (FunctionEvaluatorFactory.isGroovyExpression(transformFunction)) {
+ return;
+ }
+
validateIngestionTransformVolatility(RequestContextUtils.getExpression(transformFunction),
+ transformFunction);
+ }
+
+ private static boolean hasSameTransformConfig(@Nullable
List<TransformConfig> existingTransformConfigs,
+ TransformConfig transformConfig) {
+ if (existingTransformConfigs == null) {
+ return false;
+ }
+ for (TransformConfig existingTransformConfig : existingTransformConfigs) {
+ if (Objects.equals(existingTransformConfig.getColumnName(),
transformConfig.getColumnName())
+ && Objects.equals(existingTransformConfig.getTransformFunction(),
transformConfig.getTransformFunction())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static void validateIngestionTransformVolatility(ExpressionContext
expression,
+ String transformFunction) {
+ if (expression.getType() != ExpressionContext.Type.FUNCTION) {
+ return;
+ }
+ FunctionContext function = expression.getFunction();
+ List<ExpressionContext> arguments = function.getArguments();
+ for (ExpressionContext argument : arguments) {
+ validateIngestionTransformVolatility(argument, transformFunction);
+ }
+ String functionName = function.getFunctionName();
+ String canonicalName = FunctionRegistry.canonicalize(functionName);
+ FunctionInfo functionInfo =
FunctionRegistry.lookupFunctionInfo(canonicalName, arguments.size());
+ if (functionInfo != null && functionInfo.getVolatility() !=
FunctionVolatility.IMMUTABLE) {
+ throw new IllegalStateException(
+ String.format("Function '%s' has %s volatility and is not allowed in
ingestion transform function: %s; "
+ + "only IMMUTABLE functions are allowed", functionName,
functionInfo.getVolatility(),
+ transformFunction));
+ }
+ }
+
/// Validates all metrics-aggregation configuration for both mechanisms (the
`aggregateMetrics` flag and ingestion
/// `aggregationConfigs`) in one place, and returns the set of source
columns referenced by the aggregation functions
/// (empty when aggregation is disabled). Consuming-segment rollup keys each
row on the dictionary ids of the
@@ -912,6 +1008,11 @@ public final class TableConfigUtils {
/// - Dedup: delegates to [#validateTTLForDedupConfig]; rejects MD5 when
disabled.
@VisibleForTesting
static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema
schema) {
+ validateUpsertAndDedupConfig(tableConfig, schema, null);
+ }
+
+ static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema
schema,
+ @Nullable TableConfig existingTableConfig) {
boolean upsertEnabled = tableConfig.isUpsertEnabled();
boolean dedupEnabled = tableConfig.isDedupEnabled();
if (!upsertEnabled && !dedupEnabled) {
@@ -1039,6 +1140,8 @@ public final class TableConfigUtils {
List<TransformConfig> postPartialUpsertTransformConfigs =
upsertConfig.getPostPartialUpsertTransformConfigs();
if (postPartialUpsertTransformConfigs != null) {
+ List<TransformConfig> existingPostPartialUpsertTransformConfigs =
+ getPostPartialUpsertTransformConfigs(existingTableConfig);
Preconditions.checkState(upsertConfig.getMode() ==
UpsertConfig.Mode.PARTIAL,
"postPartialUpsertTransformConfigs can only be configured for
PARTIAL upsert tables");
Set<String> primaryKeyColumns = new
HashSet<>(schema.getPrimaryKeyColumns());
@@ -1072,6 +1175,8 @@ public final class TableConfigUtils {
+ columnName + "' in postPartialUpsertTransformConfigs");
}
try {
+ validateIngestionTransformFunctionVolatility(transformConfig,
+ existingPostPartialUpsertTransformConfigs);
FunctionEvaluator expressionEvaluator =
FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction);
List<String> arguments = expressionEvaluator.getArguments();
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/recordtransformer/ExpressionTransformerTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/recordtransformer/ExpressionTransformerTest.java
index e465874aae1..593ec59d190 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/recordtransformer/ExpressionTransformerTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/recordtransformer/ExpressionTransformerTest.java
@@ -245,6 +245,32 @@ public class ExpressionTransformerTest {
Assert.assertFalse(row.isNullValue("payload"));
}
+ @Test
+ public void
testLegacyNonDeterministicTransformFunctionRemainsRuntimeCompatible() {
+ Schema schema = new Schema.SchemaBuilder()
+ .addSingleValueDimension("eventTimeMs", FieldSpec.DataType.LONG)
+ .build();
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setTransformConfigs(List.of(new
TransformConfig("eventTimeMs", "now()")));
+ TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME)
+
.setTableName("testNonDeterministicTransformFunctionStillRunsAtRuntime")
+ .setIngestionConfig(ingestionConfig)
+ .build();
+ ExpressionTransformer expressionTransformer = new
ExpressionTransformer(tableConfig, schema);
+
+ GenericRow row = new GenericRow();
+ long lowerBound = System.currentTimeMillis();
+ expressionTransformer.transform(row);
+ long upperBound = System.currentTimeMillis();
+
+ Object value = row.getValue("eventTimeMs");
+ Assert.assertTrue(value instanceof Long, "Expected now() transform to
produce a LONG value");
+ long eventTimeMs = (Long) value;
+ long toleranceMs = 1000;
+ Assert.assertTrue(eventTimeMs >= lowerBound - toleranceMs);
+ Assert.assertTrue(eventTimeMs <= upperBound + toleranceMs);
+ }
+
/**
* If destination field already exists in the row, do not execute transform
function
*/
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
index 178b7cb8a7d..0b36f9dd52c 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
@@ -386,6 +386,60 @@ public class TableConfigUtilsTest {
ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol",
"reverse(anotherCol)")));
TableConfigUtils.validate(tableConfig, schema);
+ Schema transformSchema = schema;
+ ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol",
"now()")));
+ IllegalStateException nonDeterministicError =
+ expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validate(tableConfig, transformSchema));
+ assertTrue(nonDeterministicError.getMessage().contains("Function 'now' has
VOLATILE volatility"));
+
+ IngestionConfig existingIngestionConfig = new IngestionConfig();
+ existingIngestionConfig.setTransformConfigs(List.of(new
TransformConfig("myCol", "now()")));
+ TableConfig existingTableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setIngestionConfig(existingIngestionConfig)
+ .build();
+ TableConfigUtils.validate(tableConfig, schema, null, existingTableConfig);
+
+ ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol",
"plus(now(), 1)")));
+ nonDeterministicError = expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validate(tableConfig, transformSchema, null,
existingTableConfig));
+ assertTrue(nonDeterministicError.getMessage().contains("Function 'now' has
VOLATILE volatility"));
+
+ ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol",
"rand()")));
+ nonDeterministicError =
+ expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validate(tableConfig, transformSchema));
+ assertTrue(nonDeterministicError.getMessage().contains("Function 'rand'
has VOLATILE volatility"));
+
+ ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol",
"reqId('unused')")));
+ nonDeterministicError =
+ expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validate(tableConfig, transformSchema));
+ assertTrue(nonDeterministicError.getMessage().contains("Function 'reqid'
has STABLE volatility"),
+ nonDeterministicError.getMessage());
+
+ ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol",
"rand(123)")));
+ TableConfigUtils.validate(tableConfig, schema);
+
+ // Legacy schema-level transforms are also part of the ingestion pipeline.
A new table must reject them, while
+ // validation of an existing table stays permissive so unrelated config
updates are not stranded.
+ ingestionConfig.setTransformConfigs(List.of());
+ schema.getFieldSpecFor("myCol").setTransformFunction("now()");
+ nonDeterministicError =
+ expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validate(tableConfig, transformSchema));
+ assertTrue(nonDeterministicError.getMessage().contains("Function 'now' has
VOLATILE volatility"));
+ TableConfigUtils.validate(tableConfig, schema, null, tableConfig);
+ ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol",
"rand(123)")));
+ TableConfigUtils.validate(tableConfig, schema);
+
+ IngestionConfig noOverrideIngestionConfig = new IngestionConfig();
+ noOverrideIngestionConfig.setTransformConfigs(List.of());
+ TableConfig noOverrideTableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setIngestionConfig(noOverrideIngestionConfig)
+ .build();
+ nonDeterministicError = expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validate(noOverrideTableConfig,
transformSchema, null, tableConfig));
+ assertTrue(nonDeterministicError.getMessage().contains("Function 'now' has
VOLATILE volatility"));
+
+ schema.getFieldSpecFor("myCol").setTransformFunction(null);
+
// valid transform configs
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
.addSingleValueDimension("myCol", DataType.STRING)
@@ -4317,6 +4371,52 @@ public class TableConfigUtilsTest {
}
}
+ @Test
+ public void testPostPartialUpsertTransformVolatility() {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addSingleValueDimension("pk", DataType.STRING)
+ .addMetric("total", DataType.DOUBLE)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .setPrimaryKeyColumns(List.of("pk"))
+ .build();
+ Map<String, String> streamConfigs = getStreamConfigs();
+
+ TableConfig nonDeterministicConfig =
buildPostPartialUpsertTransformTableConfig(streamConfigs, "now()");
+ IllegalStateException nonDeterministicError =
expectThrows(IllegalStateException.class,
+ () ->
TableConfigUtils.validateUpsertAndDedupConfig(nonDeterministicConfig, schema));
+ assertTrue(nonDeterministicError.getMessage().contains("Function 'now' has
VOLATILE volatility"));
+
+ TableConfig existingConfig =
buildPostPartialUpsertTransformTableConfig(streamConfigs, "now()");
+ TableConfigUtils.validateUpsertAndDedupConfig(nonDeterministicConfig,
schema, existingConfig);
+
+ TableConfig changedConfig =
buildPostPartialUpsertTransformTableConfig(streamConfigs, "plus(now(), 1)");
+ nonDeterministicError = expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validateUpsertAndDedupConfig(changedConfig,
schema, existingConfig));
+ assertTrue(nonDeterministicError.getMessage().contains("Function 'now' has
VOLATILE volatility"));
+
+ TableConfig randomConfig =
buildPostPartialUpsertTransformTableConfig(streamConfigs, "rand()");
+ nonDeterministicError = expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validateUpsertAndDedupConfig(randomConfig,
schema));
+ assertTrue(nonDeterministicError.getMessage().contains("Function 'rand'
has VOLATILE volatility"));
+
+ TableConfig seededRandomConfig =
buildPostPartialUpsertTransformTableConfig(streamConfigs, "rand(123)");
+ TableConfigUtils.validateUpsertAndDedupConfig(seededRandomConfig, schema);
+ }
+
+ private static TableConfig
buildPostPartialUpsertTransformTableConfig(Map<String, String> streamConfigs,
+ String transformFunction) {
+ UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.PARTIAL);
+ upsertConfig.setPostPartialUpsertTransformConfigs(
+ List.of(new TransformConfig("total", transformFunction)));
+ return new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName(TIME_COLUMN)
+ .setUpsertConfig(upsertConfig)
+ .setNullHandlingEnabled(true)
+ .setStreamConfigs(streamConfigs)
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ }
+
@Test
public void
testValidateBackwardCompatibilityAllowsPartialUpsertStrategyChanges() {
// Build two PARTIAL upsert table configs that differ only in
partialUpsertStrategies and
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/annotations/FunctionVolatility.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/annotations/FunctionVolatility.java
new file mode 100644
index 00000000000..08d8f2827f8
--- /dev/null
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/annotations/FunctionVolatility.java
@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.spi.annotations;
+
+
+/// Describes how a scalar function's result or side effects can vary across
invocations.
+///
+/// The categories follow
+/// [PostgreSQL's function volatility
model](https://www.postgresql.org/docs/current/xfunc-volatility.html):
+///
+/// - `IMMUTABLE`: the result depends only on the explicit arguments and never
changes for the same inputs.
+/// - `STABLE`: the result is constant for the same arguments within one
query, but can change between queries.
+/// - `VOLATILE`: the result can change on every invocation, or the function
has side effects.
+///
+/// This semantic metadata is interpreted by context-specific policies and
does not replace
+/// [ScalarFunction#isDeterministic()], Pinot's existing compile-time query
evaluation hint.
+public enum FunctionVolatility {
+ IMMUTABLE,
+ STABLE,
+ VOLATILE
+}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/annotations/ScalarFunction.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/annotations/ScalarFunction.java
index f78b364ee57..88b573a46ff 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/annotations/ScalarFunction.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/annotations/ScalarFunction.java
@@ -62,10 +62,20 @@ public @interface ScalarFunction {
*/
boolean isVarArg() default false;
- /**
- * Whether the scalar function should be treated as deterministic (eligible
for compile-time evaluation).
- */
+ /// Whether the scalar function is eligible for compile-time query
evaluation.
+ ///
+ /// This is independent of [#volatility()] so that wall-clock functions can
retain their existing once-per-query
+ /// evaluation behavior. For backward compatibility, `false` is also treated
as [FunctionVolatility#VOLATILE].
boolean isDeterministic() default true;
+ /// Returns the semantic volatility category used by context-specific
policies.
+ ///
+ /// Persisted ingestion transforms require [FunctionVolatility#IMMUTABLE]
functions. Unlike PostgreSQL, Pinot
+ /// defaults this category to `IMMUTABLE` for compatibility with existing
UDFs, so function authors must explicitly
+ /// classify functions that read mutable or external state. This
compatibility default does not imply that every
+ /// existing function has been audited for environment-sensitive behavior.
When class-level and method-level
+ /// declarations coexist, the most volatile category wins.
+ FunctionVolatility volatility() default FunctionVolatility.IMMUTABLE;
+
@Deprecated boolean isPlaceholder() default false;
}
diff --git
a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews/fineFoodReviews_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews/fineFoodReviews_realtime_table_config.json
index 2fe1b6a0751..22d718fae0f 100644
---
a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews/fineFoodReviews_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews/fineFoodReviews_realtime_table_config.json
@@ -40,7 +40,7 @@
"transformConfigs": [
{
"columnName": "ts",
- "transformFunction": "now()"
+ "transformFunction": "1704067200000"
}
]
},
diff --git
a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_0/fineFoodReviews_part_0_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_0/fineFoodReviews_part_0_realtime_table_config.json
index dd84b3e1752..3fe48d17bbb 100644
---
a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_0/fineFoodReviews_part_0_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_0/fineFoodReviews_part_0_realtime_table_config.json
@@ -41,7 +41,7 @@
"transformConfigs": [
{
"columnName": "ts",
- "transformFunction": "now()"
+ "transformFunction": "1704067200000"
}
]
},
diff --git
a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_1/fineFoodReviews_part_1_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_1/fineFoodReviews_part_1_realtime_table_config.json
index 58e0967c856..2b2e2a47d64 100644
---
a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_1/fineFoodReviews_part_1_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_1/fineFoodReviews_part_1_realtime_table_config.json
@@ -41,7 +41,7 @@
"transformConfigs": [
{
"columnName": "ts",
- "transformFunction": "now()"
+ "transformFunction": "1704067200000"
}
]
},
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
