This is an automated email from the ASF dual-hosted git repository. xiangfu0 pushed a commit to branch data-2616-disallow-nondeterministic-transforms in repository https://gitbox.apache.org/repos/asf/pinot.git
commit dfe8f01bacf2ab3b7cd934e7eb8dd01aa95efbeb Author: Xiang Fu <[email protected]> AuthorDate: Tue Jul 7 22:36:05 2026 -0700 Disallow non-deterministic ingestion transforms Reject scalar functions annotated as non-deterministic when constructing or validating ingestion transform expressions. Mark now() and ago() as non-deterministic so table and schema transforms fail fast instead of producing inconsistent ingestion output. Keep default expression evaluator behavior permissive for existing non-ingestion callers and update SQL compiler expectations for non-deterministic functions. --- .../common/evaluator/FunctionEvaluatorFactory.java | 27 +++++- .../common/evaluator/InbuiltFunctionEvaluator.java | 9 ++ .../common/function/scalar/DateTimeFunctions.java | 6 +- .../evaluator/InbuiltFunctionEvaluatorTest.java | 22 +++++ .../pinot/sql/parsers/CalciteSqlCompilerTest.java | 102 ++++++++------------- .../pinot/core/function/FunctionRegistryTest.java | 9 ++ .../recordtransformer/ExpressionTransformer.java | 7 +- .../defaultcolumn/BaseDefaultColumnHandler.java | 3 +- .../pinot/segment/local/utils/SchemaUtils.java | 3 +- .../segment/local/utils/TableConfigUtils.java | 6 +- .../ExpressionTransformerTest.java | 28 ++++++ .../pinot/segment/local/utils/SchemaUtilsTest.java | 54 +++++++++++ .../segment/local/utils/TableConfigUtilsTest.java | 27 ++++++ 13 files changed, 224 insertions(+), 79 deletions(-) diff --git a/pinot-common/src/main/java/org/apache/pinot/common/evaluator/FunctionEvaluatorFactory.java b/pinot-common/src/main/java/org/apache/pinot/common/evaluator/FunctionEvaluatorFactory.java index 08b0374b60f..0ab47eb1c94 100644 --- a/pinot-common/src/main/java/org/apache/pinot/common/evaluator/FunctionEvaluatorFactory.java +++ b/pinot-common/src/main/java/org/apache/pinot/common/evaluator/FunctionEvaluatorFactory.java @@ -50,6 +50,16 @@ public class FunctionEvaluatorFactory { */ @Nullable public static FunctionEvaluator getExpressionEvaluator(FieldSpec fieldSpec) { + return getExpressionEvaluator(fieldSpec, true); + } + + @Nullable + public static FunctionEvaluator getDeterministicExpressionEvaluator(FieldSpec fieldSpec) { + return getExpressionEvaluator(fieldSpec, false); + } + + @Nullable + public static FunctionEvaluator getExpressionEvaluator(FieldSpec fieldSpec, boolean allowNonDeterministicFunctions) { FunctionEvaluator functionEvaluator = null; String columnName = fieldSpec.getName(); @@ -61,7 +71,7 @@ public class FunctionEvaluatorFactory { // if transform function expression present, use it to generate function evaluator try { - functionEvaluator = getExpressionEvaluator(transformExpression); + functionEvaluator = getExpressionEvaluator(transformExpression, allowNonDeterministicFunctions); } catch (Exception e) { throw new IllegalStateException( "Caught exception while constructing expression evaluator for transform expression: " + transformExpression @@ -88,22 +98,31 @@ public class FunctionEvaluatorFactory { // for backward compatible handling of Map type (currently only in Avro) String sourceMapName = columnName.substring(0, columnName.length() - MAP_KEY_COLUMN_SUFFIX.length()); String defaultMapKeysTransformExpression = getDefaultMapKeysTransformExpression(sourceMapName); - functionEvaluator = getExpressionEvaluator(defaultMapKeysTransformExpression); + functionEvaluator = getExpressionEvaluator(defaultMapKeysTransformExpression, allowNonDeterministicFunctions); } else if (columnName.endsWith(MAP_VALUE_COLUMN_SUFFIX)) { // for backward compatible handling of Map type in avro (currently only in Avro) String sourceMapName = columnName.substring(0, columnName.length() - MAP_VALUE_COLUMN_SUFFIX.length()); String defaultMapValuesTransformExpression = getDefaultMapValuesTransformExpression(sourceMapName); - functionEvaluator = getExpressionEvaluator(defaultMapValuesTransformExpression); + functionEvaluator = getExpressionEvaluator(defaultMapValuesTransformExpression, allowNonDeterministicFunctions); } return functionEvaluator; } public static FunctionEvaluator getExpressionEvaluator(String transformExpression) { + return getExpressionEvaluator(transformExpression, true); + } + + public static FunctionEvaluator getDeterministicExpressionEvaluator(String transformExpression) { + return getExpressionEvaluator(transformExpression, false); + } + + public static FunctionEvaluator getExpressionEvaluator(String transformExpression, + boolean allowNonDeterministicFunctions) { if (isGroovyExpression(transformExpression)) { return new GroovyFunctionEvaluator(transformExpression); } else { - return new InbuiltFunctionEvaluator(transformExpression); + return new InbuiltFunctionEvaluator(transformExpression, allowNonDeterministicFunctions); } } diff --git a/pinot-common/src/main/java/org/apache/pinot/common/evaluator/InbuiltFunctionEvaluator.java b/pinot-common/src/main/java/org/apache/pinot/common/evaluator/InbuiltFunctionEvaluator.java index 5db855b7327..8551e0b9314 100644 --- a/pinot-common/src/main/java/org/apache/pinot/common/evaluator/InbuiltFunctionEvaluator.java +++ b/pinot-common/src/main/java/org/apache/pinot/common/evaluator/InbuiltFunctionEvaluator.java @@ -48,9 +48,15 @@ public class InbuiltFunctionEvaluator implements FunctionEvaluator { private final ExecutableNode _rootNode; private final List<String> _arguments; private final String _functionExpression; + private final boolean _allowNonDeterministicFunctions; public InbuiltFunctionEvaluator(String functionExpression) { + this(functionExpression, true); + } + + public InbuiltFunctionEvaluator(String functionExpression, boolean allowNonDeterministicFunctions) { _functionExpression = functionExpression; + _allowNonDeterministicFunctions = allowNonDeterministicFunctions; _arguments = new ArrayList<>(); _rootNode = planExecution(RequestContextUtils.getExpression(functionExpression)); } @@ -99,6 +105,9 @@ public class InbuiltFunctionEvaluator implements FunctionEvaluator { throw new IllegalStateException(String.format("Unsupported function: %s", functionName)); } } + Preconditions.checkState(_allowNonDeterministicFunctions || functionInfo.isDeterministic(), + "Non-deterministic function '%s' is not allowed in deterministic-only expression: %s", functionName, + _functionExpression); return new FunctionExecutionNode(functionInfo, childNodes); } default: 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..b1c442736cf 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 @@ -573,7 +573,7 @@ public class DateTimeFunctions { * Return current time as epoch millis * TODO: Consider changing the return type to Timestamp */ - @ScalarFunction + @ScalarFunction(isDeterministic = false) public static long now() { return System.currentTimeMillis(); } @@ -603,13 +603,13 @@ public class DateTimeFunctions { * "-P6H3M" -- parses as "-6 hours and -3 minutes" * "-P-6H+3M" -- parses as "+6 hours and -3 minutes" */ - @ScalarFunction + @ScalarFunction(isDeterministic = false) public static long ago(String periodString) { Duration period = Duration.parse(periodString); return System.currentTimeMillis() - period.toMillis(); } - @ScalarFunction + @ScalarFunction(isDeterministic = false) 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/test/java/org/apache/pinot/common/evaluator/InbuiltFunctionEvaluatorTest.java b/pinot-common/src/test/java/org/apache/pinot/common/evaluator/InbuiltFunctionEvaluatorTest.java index 116517a6c79..ef877bc99e2 100644 --- a/pinot-common/src/test/java/org/apache/pinot/common/evaluator/InbuiltFunctionEvaluatorTest.java +++ b/pinot-common/src/test/java/org/apache/pinot/common/evaluator/InbuiltFunctionEvaluatorTest.java @@ -29,6 +29,7 @@ import org.testng.annotations.Test; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNull; import static org.testng.Assert.assertTrue; +import static org.testng.Assert.expectThrows; public class InbuiltFunctionEvaluatorTest { @@ -250,4 +251,25 @@ public class InbuiltFunctionEvaluatorTest { longRow.putValue("shift", 40); assertEquals(new InbuiltFunctionEvaluator("bitExtract(value, shift)").evaluate(longRow), 1); } + + @Test + public void testNonDeterministicFunctionsRejectedInDeterministicOnlyMode() { + assertTrue(new InbuiltFunctionEvaluator("now()").evaluate(new GenericRow()) instanceof Long); + + assertEquals(new InbuiltFunctionEvaluator("rand(123)", false).evaluate(new GenericRow()), + new InbuiltFunctionEvaluator("rand(123)", false).evaluate(new GenericRow())); + + IllegalStateException e = expectThrows(IllegalStateException.class, + () -> new InbuiltFunctionEvaluator("now()", false)); + assertTrue(e.getMessage().contains("Non-deterministic function 'now'")); + + e = expectThrows(IllegalStateException.class, () -> new InbuiltFunctionEvaluator("ago('PT1H')", false)); + assertTrue(e.getMessage().contains("Non-deterministic function 'ago'")); + + e = expectThrows(IllegalStateException.class, () -> new InbuiltFunctionEvaluator("rand()", false)); + assertTrue(e.getMessage().contains("Non-deterministic function 'rand'")); + + e = expectThrows(IllegalStateException.class, () -> new InbuiltFunctionEvaluator("plus(now(), 1)", false)); + assertTrue(e.getMessage().contains("Non-deterministic function 'now'")); + } } diff --git a/pinot-common/src/test/java/org/apache/pinot/sql/parsers/CalciteSqlCompilerTest.java b/pinot-common/src/test/java/org/apache/pinot/sql/parsers/CalciteSqlCompilerTest.java index 9db25530264..d6dc2800bd0 100644 --- a/pinot-common/src/test/java/org/apache/pinot/sql/parsers/CalciteSqlCompilerTest.java +++ b/pinot-common/src/test/java/org/apache/pinot/sql/parsers/CalciteSqlCompilerTest.java @@ -18,12 +18,8 @@ */ package org.apache.pinot.sql.parsers; -import java.time.Instant; -import java.time.ZoneId; -import java.time.format.DateTimeFormatter; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.TimeUnit; import org.apache.calcite.sql.SqlNumericLiteral; import org.apache.pinot.common.request.DataSource; import org.apache.pinot.common.request.Expression; @@ -47,8 +43,6 @@ import org.testng.annotations.Test; * Some tests for the SQL compiler. */ public class CalciteSqlCompilerTest { - private static final long ONE_HOUR_IN_MS = TimeUnit.HOURS.toMillis(1); - /* Verify all lists in PinotQuery are ArrayLists because we might need to modify them during query optimization */ private Expression compileToExpression(String expressionStr) { @@ -241,8 +235,9 @@ public class CalciteSqlCompilerTest { public void testCaseWhenScalar() { PinotQuery pinotQuery = compileToPinotQuery("SELECT CASE WHEN NOW() > 0 THEN 1 ELSE -1 END FROM myTable"); Assert.assertEquals(pinotQuery.getSelectList().size(), 1); - Assert.assertTrue(pinotQuery.getSelectList().get(0).isSetLiteral()); - Assert.assertEquals(pinotQuery.getSelectList().get(0).getLiteral().getIntValue(), 1); + Expression caseExpression = pinotQuery.getSelectList().get(0); + Assert.assertTrue(caseExpression.isSetFunctionCall()); + Assert.assertEquals(caseExpression.getFunctionCall().getOperator(), "case"); Assert.assertThrows(SqlCompilationException.class, () -> compileToPinotQuery("SELECT CASE WHEN 1 > 0 END FROM myTable")); @@ -2376,44 +2371,42 @@ public class CalciteSqlCompilerTest { @Test public void testCompilationInvokedFunction() { String query = "SELECT now() FROM foo"; - long lowerBound = System.currentTimeMillis(); PinotQuery pinotQuery = compileToPinotQuery(query); - long nowTs = pinotQuery.getSelectList().get(0).getLiteral().getLongValue(); - long upperBound = System.currentTimeMillis(); - Assert.assertTrue(nowTs >= lowerBound); - Assert.assertTrue(nowTs <= upperBound); + Expression nowExpression = pinotQuery.getSelectList().get(0); + Assert.assertTrue(nowExpression.isSetFunctionCall()); + Assert.assertEquals(nowExpression.getFunctionCall().getOperator(), "now"); + Assert.assertTrue(nowExpression.getFunctionCall().getOperands().isEmpty()); query = "SELECT a FROM foo where time_col > now()"; - lowerBound = System.currentTimeMillis(); pinotQuery = compileToPinotQuery(query); Function greaterThan = pinotQuery.getFilterExpression().getFunctionCall(); - nowTs = greaterThan.getOperands().get(1).getLiteral().getLongValue(); - upperBound = System.currentTimeMillis(); - Assert.assertTrue(nowTs >= lowerBound); - Assert.assertTrue(nowTs <= upperBound); + Function minus = greaterThan.getOperands().get(0).getFunctionCall(); + nowExpression = minus.getOperands().get(1); + Assert.assertTrue(nowExpression.isSetFunctionCall()); + Assert.assertEquals(nowExpression.getFunctionCall().getOperator(), "now"); + Assert.assertTrue(nowExpression.getFunctionCall().getOperands().isEmpty()); query = "SELECT a FROM foo where time_col > fromDateTime('2020-01-01 UTC', 'yyyy-MM-dd z')"; pinotQuery = compileToPinotQuery(query); greaterThan = pinotQuery.getFilterExpression().getFunctionCall(); - nowTs = greaterThan.getOperands().get(1).getLiteral().getLongValue(); + long nowTs = greaterThan.getOperands().get(1).getLiteral().getLongValue(); Assert.assertEquals(nowTs, 1577836800000L); query = "SELECT ago('PT1H') FROM foo"; - lowerBound = System.currentTimeMillis() - ONE_HOUR_IN_MS; pinotQuery = compileToPinotQuery(query); - nowTs = pinotQuery.getSelectList().get(0).getLiteral().getLongValue(); - upperBound = System.currentTimeMillis() - ONE_HOUR_IN_MS; - Assert.assertTrue(nowTs >= lowerBound); - Assert.assertTrue(nowTs <= upperBound); + Expression agoExpression = pinotQuery.getSelectList().get(0); + Assert.assertTrue(agoExpression.isSetFunctionCall()); + Assert.assertEquals(agoExpression.getFunctionCall().getOperator(), "ago"); + Assert.assertEquals(agoExpression.getFunctionCall().getOperands().get(0).getLiteral().getStringValue(), "PT1H"); query = "SELECT a FROM foo where time_col > ago('PT1H')"; - lowerBound = System.currentTimeMillis() - ONE_HOUR_IN_MS; pinotQuery = compileToPinotQuery(query); greaterThan = pinotQuery.getFilterExpression().getFunctionCall(); - nowTs = greaterThan.getOperands().get(1).getLiteral().getLongValue(); - upperBound = System.currentTimeMillis() - ONE_HOUR_IN_MS; - Assert.assertTrue(nowTs >= lowerBound); - Assert.assertTrue(nowTs <= upperBound); + minus = greaterThan.getOperands().get(0).getFunctionCall(); + agoExpression = minus.getOperands().get(1); + Assert.assertTrue(agoExpression.isSetFunctionCall()); + Assert.assertEquals(agoExpression.getFunctionCall().getOperator(), "ago"); + Assert.assertEquals(agoExpression.getFunctionCall().getOperands().get(0).getLiteral().getStringValue(), "PT1H"); query = "SELECT rand() FROM foo"; pinotQuery = compileToPinotQuery(query); @@ -2647,72 +2640,53 @@ public class CalciteSqlCompilerTest { String query = "SELECT a FROM foo where time_col > toDateTime(now(), 'yyyy-MM-dd z')"; PinotQuery pinotQuery = compileToPinotQuery(query); Function greaterThan = pinotQuery.getFilterExpression().getFunctionCall(); - String today = greaterThan.getOperands().get(1).getLiteral().getStringValue(); - String expectedTodayStr = - Instant.now().atZone(ZoneId.of("UTC")).format(DateTimeFormatter.ofPattern("yyyy-MM-dd z")); - Assert.assertEquals(today, expectedTodayStr); + Expression toDateTimeExpression = greaterThan.getOperands().get(1); + Assert.assertTrue(toDateTimeExpression.isSetLiteral()); } @Test public void testCompileTimeExpression() { - long lowerBound = System.currentTimeMillis(); Expression expression = compileToExpression("now()"); Assert.assertNotNull(expression.getFunctionCall()); expression = CompileTimeFunctionsInvoker.invokeCompileTimeFunctionExpression(expression); - Assert.assertNotNull(expression.getLiteral()); - long upperBound = System.currentTimeMillis(); - long result = expression.getLiteral().getLongValue(); - Assert.assertTrue(result >= lowerBound && result <= upperBound); + Assert.assertNotNull(expression.getFunctionCall()); + Assert.assertFalse(expression.isSetLiteral()); expression = compileToExpression("now() - 0"); Assert.assertNotNull(expression.getFunctionCall()); expression = CompileTimeFunctionsInvoker.invokeCompileTimeFunctionExpression(expression); - Assert.assertNotNull(expression.getLiteral()); - upperBound = System.currentTimeMillis(); - result = expression.getLiteral().getLongValue(); - Assert.assertTrue(result >= lowerBound && result <= upperBound); + Assert.assertNotNull(expression.getFunctionCall()); + Assert.assertFalse(expression.isSetLiteral()); expression = compileToExpression("now() + 0"); Assert.assertNotNull(expression.getFunctionCall()); expression = CompileTimeFunctionsInvoker.invokeCompileTimeFunctionExpression(expression); - Assert.assertNotNull(expression.getLiteral()); - upperBound = System.currentTimeMillis(); - result = expression.getLiteral().getLongValue(); - Assert.assertTrue(result >= lowerBound && result <= upperBound); + Assert.assertNotNull(expression.getFunctionCall()); + Assert.assertFalse(expression.isSetLiteral()); expression = compileToExpression("now() * 1"); Assert.assertNotNull(expression.getFunctionCall()); expression = CompileTimeFunctionsInvoker.invokeCompileTimeFunctionExpression(expression); - Assert.assertNotNull(expression.getLiteral()); - upperBound = System.currentTimeMillis(); - result = expression.getLiteral().getLongValue(); - Assert.assertTrue(result >= lowerBound && result <= upperBound); + Assert.assertNotNull(expression.getFunctionCall()); + Assert.assertFalse(expression.isSetLiteral()); - lowerBound = TimeUnit.MILLISECONDS.toHours(System.currentTimeMillis()) + 1; expression = compileToExpression("to_epoch_hours(now() + 3600000)"); Assert.assertNotNull(expression.getFunctionCall()); expression = CompileTimeFunctionsInvoker.invokeCompileTimeFunctionExpression(expression); - upperBound = TimeUnit.MILLISECONDS.toHours(System.currentTimeMillis()) + 1; - result = expression.getLiteral().getLongValue(); - Assert.assertTrue(result >= lowerBound && result <= upperBound); + Assert.assertNotNull(expression.getFunctionCall()); + Assert.assertFalse(expression.isSetLiteral()); - lowerBound = System.currentTimeMillis() - ONE_HOUR_IN_MS; expression = compileToExpression("ago('PT1H')"); Assert.assertNotNull(expression.getFunctionCall()); expression = CompileTimeFunctionsInvoker.invokeCompileTimeFunctionExpression(expression); - Assert.assertNotNull(expression.getLiteral()); - upperBound = System.currentTimeMillis() - ONE_HOUR_IN_MS; - result = expression.getLiteral().getLongValue(); - Assert.assertTrue(result >= lowerBound && result <= upperBound); + Assert.assertNotNull(expression.getFunctionCall()); + Assert.assertFalse(expression.isSetLiteral()); - lowerBound = System.currentTimeMillis() + ONE_HOUR_IN_MS; expression = compileToExpression("ago('PT-1H')"); Assert.assertNotNull(expression.getFunctionCall()); expression = CompileTimeFunctionsInvoker.invokeCompileTimeFunctionExpression(expression); - Assert.assertNotNull(expression.getLiteral()); - upperBound = System.currentTimeMillis() + ONE_HOUR_IN_MS; - result = expression.getLiteral().getLongValue(); - Assert.assertTrue(result >= lowerBound && result <= upperBound); + Assert.assertNotNull(expression.getFunctionCall()); + Assert.assertFalse(expression.isSetLiteral()); expression = compileToExpression("toDateTime(millisSinceEpoch)"); Assert.assertNotNull(expression.getFunctionCall()); diff --git a/pinot-core/src/test/java/org/apache/pinot/core/function/FunctionRegistryTest.java b/pinot-core/src/test/java/org/apache/pinot/core/function/FunctionRegistryTest.java index abc996d2447..913d91c504e 100644 --- a/pinot-core/src/test/java/org/apache/pinot/core/function/FunctionRegistryTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/core/function/FunctionRegistryTest.java @@ -29,6 +29,7 @@ import org.apache.pinot.sql.FilterKind; 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; @@ -95,6 +96,14 @@ public class FunctionRegistryTest { assertNull(FunctionRegistry.lookupFunctionInfo("testfunc2", 1)); } + @Test + public void testScalarFunctionDeterminism() { + assertFalse(FunctionRegistry.lookupFunctionInfo("now", 0).isDeterministic()); + assertFalse(FunctionRegistry.lookupFunctionInfo("ago", 1).isDeterministic()); + assertFalse(FunctionRegistry.lookupFunctionInfo("rand", 0).isDeterministic()); + assertTrue(FunctionRegistry.lookupFunctionInfo("rand", 1).isDeterministic()); + } + @Test public void testBitFunctionPolymorphism() { FunctionInfo intBitAnd = diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/ExpressionTransformer.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/ExpressionTransformer.java index b7ac7ecbf41..f68333e85b0 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/ExpressionTransformer.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/ExpressionTransformer.java @@ -75,7 +75,7 @@ public class ExpressionTransformer implements RecordTransformer { if (transformConfigs != null) { for (TransformConfig transformConfig : transformConfigs) { FunctionEvaluator previous = expressionEvaluators.put(transformConfig.getColumnName(), - FunctionEvaluatorFactory.getExpressionEvaluator(transformConfig.getTransformFunction())); + FunctionEvaluatorFactory.getDeterministicExpressionEvaluator(transformConfig.getTransformFunction())); Preconditions.checkState(previous == null, "Cannot set more than one transform function on column: %s.", transformConfig.getColumnName()); } @@ -83,7 +83,8 @@ public class ExpressionTransformer implements RecordTransformer { for (FieldSpec fieldSpec : schema.getAllFieldSpecs()) { String fieldName = fieldSpec.getName(); if (!fieldSpec.isVirtualColumn() && !expressionEvaluators.containsKey(fieldName)) { - FunctionEvaluator functionEvaluator = FunctionEvaluatorFactory.getExpressionEvaluator(fieldSpec); + FunctionEvaluator functionEvaluator = + FunctionEvaluatorFactory.getDeterministicExpressionEvaluator(fieldSpec); if (functionEvaluator != null) { expressionEvaluators.put(fieldName, functionEvaluator); if (isImplicitMapTransform(fieldSpec)) { @@ -108,7 +109,7 @@ public class ExpressionTransformer implements RecordTransformer { Map<String, FunctionEvaluator> expressionEvaluators = new HashMap<>(); for (TransformConfig transformConfig : transformConfigs) { FunctionEvaluator previous = expressionEvaluators.put(transformConfig.getColumnName(), - FunctionEvaluatorFactory.getExpressionEvaluator(transformConfig.getTransformFunction())); + FunctionEvaluatorFactory.getDeterministicExpressionEvaluator(transformConfig.getTransformFunction())); Preconditions.checkState(previous == null, "Cannot set more than one transform function on column: %s.", transformConfig.getColumnName()); } diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java index d88b3b90c60..f03a20469c4 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java @@ -396,7 +396,8 @@ public abstract class BaseDefaultColumnHandler implements DefaultColumnHandler { for (TransformConfig transformConfig : transformConfigs) { if (transformConfig.getColumnName().equals(column)) { String transformFunction = transformConfig.getTransformFunction(); - FunctionEvaluator functionEvaluator = FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction); + FunctionEvaluator functionEvaluator = + FunctionEvaluatorFactory.getDeterministicExpressionEvaluator(transformFunction); // Check if all arguments exist in the segment // TODO: Support chained derived column 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 b4ae42641c0..116c62efbcf 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 @@ -139,7 +139,8 @@ public class SchemaUtils { String transformFunction = fieldSpec.getTransformFunction(); if (transformFunction != null) { try { - List<String> arguments = FunctionEvaluatorFactory.getExpressionEvaluator(fieldSpec).getArguments(); + List<String> arguments = + FunctionEvaluatorFactory.getDeterministicExpressionEvaluator(fieldSpec).getArguments(); Preconditions.checkState(!arguments.contains(column), "The arguments of transform function %s should not contain the destination column %s", transformFunction, column); 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 c833b7cdaa3..37786aec007 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 @@ -675,7 +675,7 @@ public final class TableConfigUtils { && !(_disableGroovy && FunctionEvaluatorFactory.isGroovyExpression(transformFunction))) { try { transformInputColumns.addAll( - FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction).getArguments()); + FunctionEvaluatorFactory.getDeterministicExpressionEvaluator(transformFunction).getArguments()); } catch (Exception ignore) { // Invalid functions are reported with a descriptive error in the main loop below. } @@ -705,7 +705,7 @@ public final class TableConfigUtils { + "' for column '" + columnName + "'"); } try { - expressionEvaluator = FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction); + expressionEvaluator = FunctionEvaluatorFactory.getDeterministicExpressionEvaluator(transformFunction); } catch (Exception e) { throw new IllegalStateException( "Invalid transform function '" + transformFunction + "' for column '" + columnName + "', exception: " @@ -1013,7 +1013,7 @@ public final class TableConfigUtils { } try { FunctionEvaluator expressionEvaluator = - FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction); + FunctionEvaluatorFactory.getDeterministicExpressionEvaluator(transformFunction); List<String> arguments = expressionEvaluator.getArguments(); if (arguments.contains(columnName)) { throw new IllegalStateException( 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 5f00033421e..512df6ac687 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 @@ -384,6 +384,34 @@ public class ExpressionTransformerTest { Assert.assertEquals(record.getValue("f"), 210.0); } + @Test + public void testRejectNonDeterministicTransformFunction() { + Schema schema = new Schema.SchemaBuilder().addSingleValueDimension("createdAt", FieldSpec.DataType.LONG).build(); + IngestionConfig ingestionConfig = new IngestionConfig(); + ingestionConfig.setTransformConfigs(Collections.singletonList(new TransformConfig("createdAt", "now()"))); + TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME) + .setTableName("testRejectNonDeterministicTransformFunction") + .setIngestionConfig(ingestionConfig) + .build(); + + IllegalStateException e = + Assert.expectThrows(IllegalStateException.class, () -> new ExpressionTransformer(tableConfig, schema)); + Assert.assertTrue(e.getMessage().contains("Non-deterministic function 'now'")); + } + + @Test + public void testRejectNonDeterministicSchemaTransformFunction() { + Schema schema = new Schema.SchemaBuilder().addSingleValueDimension("createdAt", FieldSpec.DataType.LONG).build(); + schema.getFieldSpecFor("createdAt").setTransformFunction("ago('PT1H')"); + TableConfig tableConfig = + new TableConfigBuilder(TableType.REALTIME).setTableName("testRejectNonDeterministicSchemaTransformFunction") + .setIngestionConfig(new IngestionConfig()).build(); + + IllegalStateException e = + Assert.expectThrows(IllegalStateException.class, () -> new ExpressionTransformer(tableConfig, schema)); + Assert.assertTrue(e.getMessage().contains("Non-deterministic function 'ago'")); + } + /** Check if there is more than one transform function definition for the same column. */ @Test(expectedExceptions = IllegalStateException.class, expectedExceptionsMessageRegExp = "Cannot set more than one" + " transform function on column: a.") diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/SchemaUtilsTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/SchemaUtilsTest.java new file mode 100644 index 00000000000..d73fd880727 --- /dev/null +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/SchemaUtilsTest.java @@ -0,0 +1,54 @@ +/** + * 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.segment.local.utils; + +import org.apache.pinot.spi.data.FieldSpec; +import org.apache.pinot.spi.data.Schema; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.expectThrows; + + +/// Tests [SchemaUtils] validation. +public class SchemaUtilsTest { + private static final String SCHEMA_NAME = "testSchema"; + + @Test + public void testRejectNonDeterministicSchemaTransformFunction() { + Schema schema = new Schema.SchemaBuilder().setSchemaName(SCHEMA_NAME) + .addSingleValueDimension("createdAt", FieldSpec.DataType.LONG) + .build(); + schema.getFieldSpecFor("createdAt").setTransformFunction("ago('PT1H')"); + + IllegalStateException error = expectThrows(IllegalStateException.class, () -> SchemaUtils.validate(schema)); + assertTrue(error.getMessage().contains("Exception in getting arguments for transform function 'ago('PT1H')'")); + assertTrue(error.getCause().getMessage().contains("Non-deterministic function 'ago'")); + } + + @Test + public void testAllowDeterministicSchemaTransformFunction() { + Schema schema = new Schema.SchemaBuilder().setSchemaName(SCHEMA_NAME) + .addSingleValueDimension("bucket", FieldSpec.DataType.LONG) + .build(); + schema.getFieldSpecFor("bucket").setTransformFunction("toEpochHours(timestampColumn)"); + + SchemaUtils.validate(schema); + } +} 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 0776eeda342..64059310f09 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,20 @@ public class TableConfigUtilsTest { ingestionConfig.setTransformConfigs(Collections.singletonList(new TransformConfig("myCol", "reverse(anotherCol)"))); TableConfigUtils.validate(tableConfig, schema); + Schema transformSchema = schema; + ingestionConfig.setTransformConfigs(Collections.singletonList(new TransformConfig("myCol", "now()"))); + IllegalStateException nonDeterministicError = + expectThrows(IllegalStateException.class, () -> TableConfigUtils.validate(tableConfig, transformSchema)); + assertTrue(nonDeterministicError.getMessage().contains("Non-deterministic function 'now'")); + + ingestionConfig.setTransformConfigs(Collections.singletonList(new TransformConfig("myCol", "rand()"))); + nonDeterministicError = + expectThrows(IllegalStateException.class, () -> TableConfigUtils.validate(tableConfig, transformSchema)); + assertTrue(nonDeterministicError.getMessage().contains("Non-deterministic function 'rand'")); + + ingestionConfig.setTransformConfigs(Collections.singletonList(new TransformConfig("myCol", "rand(123)"))); + TableConfigUtils.validate(tableConfig, schema); + // valid transform configs schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME) .addSingleValueDimension("myCol", FieldSpec.DataType.STRING) @@ -4076,6 +4090,19 @@ public class TableConfigUtilsTest { .build(); TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema); + UpsertConfig nonDeterministicConfig = new UpsertConfig(UpsertConfig.Mode.PARTIAL); + nonDeterministicConfig.setPostPartialUpsertTransformConfigs(List.of(new TransformConfig("total", "now()"))); + TableConfig nonDeterministicTableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME) + .setTimeColumnName(TIME_COLUMN) + .setUpsertConfig(nonDeterministicConfig) + .setStreamConfigs(streamConfigs) + .setRoutingConfig( + new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false)) + .build(); + IllegalStateException nonDeterministicError = expectThrows(IllegalStateException.class, + () -> TableConfigUtils.validateUpsertAndDedupConfig(nonDeterministicTableConfig, schema)); + assertTrue(nonDeterministicError.getMessage().contains("Non-deterministic function 'now'")); + // postPartialUpsertTransformConfigs on FULL upsert should fail UpsertConfig fullUpsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL); fullUpsertConfig.setPostPartialUpsertTransformConfigs( --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
