From 65a9fd185b441b1d0a0a4e3ff5f053f97ec366a8 Mon Sep 17 00:00:00 2001 From: Xiang Fu Date: Tue, 7 Jul 2026 22:36:05 -0700 Subject: [PATCH] Disallow non-deterministic ingestion transform configs Reject non-deterministic built-in functions during table config validation for new or changed ingestion transform configs and post-partial-upsert transform configs. Keep runtime expression evaluation permissive, and allow unchanged legacy transform configs during table updates so existing tables continue ingesting and can receive unrelated config edits. --- .../LiteralOnlyBrokerRequestTest.java | 53 +++------ .../function/scalar/DateTimeFunctions.java | 37 +++--- .../common/function/FunctionUtilsTest.java | 8 ++ .../sql/parsers/CalciteSqlCompilerTest.java | 112 +++++++----------- .../resources/PinotDdlRestletResource.java | 2 +- .../resources/PinotTableRestletResource.java | 7 +- .../resources/TableConfigValidationUtils.java | 15 +-- .../TableConfigsRestletResource.java | 6 +- ...stletResourceMaterializedViewUnitTest.java | 2 +- .../tests/custom/CLPEncodingRealtimeTest.java | 2 +- .../query/validate/PinotTypeCoercionTest.java | 54 +++------ .../segment/local/utils/TableConfigUtils.java | 87 +++++++++++++- .../ExpressionTransformerTest.java | 26 ++++ .../local/utils/TableConfigUtilsTest.java | 96 +++++++++------ ...fineFoodReviews_realtime_table_config.json | 2 +- ...dReviews_part_0_realtime_table_config.json | 2 +- ...dReviews_part_1_realtime_table_config.json | 2 +- 17 files changed, 292 insertions(+), 221 deletions(-) diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/LiteralOnlyBrokerRequestTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/LiteralOnlyBrokerRequestTest.java index a40a1e9cf816..0db10d6a1742 100644 --- a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/LiteralOnlyBrokerRequestTest.java +++ b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/LiteralOnlyBrokerRequestTest.java @@ -20,7 +20,6 @@ import java.util.List; import java.util.Random; -import java.util.concurrent.TimeUnit; import org.apache.pinot.broker.broker.AccessControlFactory; import org.apache.pinot.broker.broker.AllowAllAccessControlFactory; import org.apache.pinot.common.failuredetector.FailureDetector; @@ -48,7 +47,6 @@ public class LiteralOnlyBrokerRequestTest { private static final AccessControlFactory ACCESS_CONTROL_FACTORY = new AllowAllAccessControlFactory(); private static final Random RANDOM = new Random(System.currentTimeMillis()); - private static final long ONE_HOUR_IN_MS = TimeUnit.HOURS.toMillis(1); @BeforeClass public void setUp() { @@ -82,18 +80,20 @@ public void testNumberLiteralBrokerRequestFromSQL() { @Test public void testLiteralOnlyTransformBrokerRequestFromSQL() { - assertTrue(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery("SELECT now()"))); - assertTrue(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery("SELECT ago('PT1H')"))); - assertTrue(isLiteralOnlyQuery( + assertFalse(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery("SELECT now()"))); + assertFalse(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery("SELECT ago('PT1H')"))); + assertFalse(isLiteralOnlyQuery( CalciteSqlParser.compileToPinotQuery("SELECT now(), fromDateTime('2020-01-01 UTC', 'yyyy-MM-dd z')"))); - assertTrue(isLiteralOnlyQuery( + assertFalse(isLiteralOnlyQuery( CalciteSqlParser.compileToPinotQuery("SELECT ago('PT1H'), fromDateTime('2020-01-01 UTC', 'yyyy-MM-dd z')"))); - assertTrue(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery("SELECT now() FROM myTable"))); - assertTrue(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery("SELECT ago('PT1H') FROM myTable"))); - assertTrue(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery( + assertFalse(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery("SELECT now() FROM myTable"))); + assertFalse(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery("SELECT ago('PT1H') FROM myTable"))); + assertFalse(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery( "SELECT now(), fromDateTime('2020-01-01 UTC', 'yyyy-MM-dd z') FROM myTable"))); - assertTrue(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery( + assertFalse(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery( "SELECT ago('PT1H'), fromDateTime('2020-01-01 UTC', 'yyyy-MM-dd z') FROM myTable"))); + assertTrue(isLiteralOnlyQuery( + CalciteSqlParser.compileToPinotQuery("SELECT fromDateTime('2020-01-01 UTC', 'yyyy-MM-dd z')"))); assertFalse( isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery("SELECT count(*) from foo where bar > ago('PT1H')"))); assertTrue(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery( @@ -152,9 +152,9 @@ public void testLiteralOnlyTransformBrokerRequestFromSQL() { @Test public void testLiteralOnlyWithAsBrokerRequestFromSQL() { - assertTrue(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery( + assertFalse(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery( "SELECT now() AS currentTs, fromDateTime('2020-01-01 UTC', 'yyyy-MM-dd z') AS firstDayOf2020"))); - assertTrue(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery( + assertFalse(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery( "SELECT ago('PT1H') AS currentTs, fromDateTime('2020-01-01 UTC', 'yyyy-MM-dd z') AS firstDayOf2020"))); assertTrue(isLiteralOnlyQuery(CalciteSqlParser.compileToPinotQuery( "SELECT encodeUrl('key1=value 1&key2=value@!$2&key3=value%3') AS encoded, " @@ -201,34 +201,13 @@ public void testBrokerRequestHandlerWithAsFunction() new BrokerRequestIdGenerator(), null, ACCESS_CONTROL_FACTORY, null, null, null, null, mock(ServerRoutingStatsManager.class), mock(FailureDetector.class), ThreadAccountantUtils.getNoOpAccountant(), null, null); - long currentTsMin = System.currentTimeMillis(); BrokerResponse brokerResponse = requestHandler.handleRequest( - "SELECT now() AS currentTs, fromDateTime('2020-01-01 UTC', 'yyyy-MM-dd z') AS firstDayOf2020"); - long currentTsMax = System.currentTimeMillis(); - assertEquals(brokerResponse.getResultTable().getDataSchema().getColumnName(0), "currentTs"); - assertEquals(brokerResponse.getResultTable().getDataSchema().getColumnDataType(0), DataSchema.ColumnDataType.LONG); - assertEquals(brokerResponse.getResultTable().getDataSchema().getColumnName(1), "firstDayOf2020"); - assertEquals(brokerResponse.getResultTable().getDataSchema().getColumnDataType(1), DataSchema.ColumnDataType.LONG); - assertEquals(brokerResponse.getResultTable().getRows().size(), 1); - assertEquals(brokerResponse.getResultTable().getRows().get(0).length, 2); - assertTrue(Long.parseLong(brokerResponse.getResultTable().getRows().get(0)[0].toString()) > currentTsMin); - assertTrue(Long.parseLong(brokerResponse.getResultTable().getRows().get(0)[0].toString()) < currentTsMax); - assertEquals(brokerResponse.getResultTable().getRows().get(0)[1], 1577836800000L); - assertEquals(brokerResponse.getTotalDocs(), 0); - - long oneHourAgoTsMin = System.currentTimeMillis() - ONE_HOUR_IN_MS; - brokerResponse = requestHandler.handleRequest( - "SELECT ago('PT1H') AS oneHourAgoTs, fromDateTime('2020-01-01 UTC', 'yyyy-MM-dd z') AS firstDayOf2020"); - long oneHourAgoTsMax = System.currentTimeMillis() - ONE_HOUR_IN_MS; - assertEquals(brokerResponse.getResultTable().getDataSchema().getColumnName(0), "oneHourAgoTs"); + "SELECT fromDateTime('2020-01-01 UTC', 'yyyy-MM-dd z') AS firstDayOf2020"); + assertEquals(brokerResponse.getResultTable().getDataSchema().getColumnName(0), "firstDayOf2020"); assertEquals(brokerResponse.getResultTable().getDataSchema().getColumnDataType(0), DataSchema.ColumnDataType.LONG); - assertEquals(brokerResponse.getResultTable().getDataSchema().getColumnName(1), "firstDayOf2020"); - assertEquals(brokerResponse.getResultTable().getDataSchema().getColumnDataType(1), DataSchema.ColumnDataType.LONG); assertEquals(brokerResponse.getResultTable().getRows().size(), 1); - assertEquals(brokerResponse.getResultTable().getRows().get(0).length, 2); - assertTrue(Long.parseLong(brokerResponse.getResultTable().getRows().get(0)[0].toString()) >= oneHourAgoTsMin); - assertTrue(Long.parseLong(brokerResponse.getResultTable().getRows().get(0)[0].toString()) <= oneHourAgoTsMax); - assertEquals(brokerResponse.getResultTable().getRows().get(0)[1], 1577836800000L); + assertEquals(brokerResponse.getResultTable().getRows().get(0).length, 1); + assertEquals(brokerResponse.getResultTable().getRows().get(0)[0], 1577836800000L); assertEquals(brokerResponse.getTotalDocs(), 0); brokerResponse = requestHandler.handleRequest( 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 4da99e84cb3d..75cdc9842dd1 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 @@ -569,11 +569,10 @@ public static long[] roundMV(long[] timeValue, long roundToNearest) { return results; } - /** - * Return current time as epoch millis - * TODO: Consider changing the return type to Timestamp - */ - @ScalarFunction + /// Returns current time as epoch millis. + /// + /// TODO: Consider changing the return type to Timestamp. + @ScalarFunction(isDeterministic = false) public static long now() { return System.currentTimeMillis(); } @@ -591,25 +590,25 @@ public static long sleep(long millis) { return millis; } - /** - * Return time as epoch millis before the given period (in ISO-8601 duration format). - * Examples: - * "PT20.345S" -- parses as "20.345 seconds" - * "PT15M" -- parses as "15 minutes" (where a minute is 60 seconds) - * "PT10H" -- parses as "10 hours" (where an hour is 3600 seconds) - * "P2D" -- parses as "2 days" (where a day is 24 hours or 86400 seconds) - * "P2DT3H4M" -- parses as "2 days, 3 hours and 4 minutes" - * "P-6H3M" -- parses as "-6 hours and +3 minutes" - * "-P6H3M" -- parses as "-6 hours and -3 minutes" - * "-P-6H+3M" -- parses as "+6 hours and -3 minutes" - */ - @ScalarFunction + /// Returns time as epoch millis before the given period (in ISO-8601 duration format). + /// + /// Examples: + /// + /// - `"PT20.345S"` parses as "20.345 seconds". + /// - `"PT15M"` parses as "15 minutes" (where a minute is 60 seconds). + /// - `"PT10H"` parses as "10 hours" (where an hour is 3600 seconds). + /// - `"P2D"` parses as "2 days" (where a day is 24 hours or 86400 seconds). + /// - `"P2DT3H4M"` parses as "2 days, 3 hours and 4 minutes". + /// - `"P-6H3M"` parses as "-6 hours and +3 minutes". + /// - `"-P6H3M"` parses as "-6 hours and -3 minutes". + /// - `"-P-6H+3M"` parses as "+6 hours and -3 minutes". + @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/function/FunctionUtilsTest.java b/pinot-common/src/test/java/org/apache/pinot/common/function/FunctionUtilsTest.java index ef87afa66879..b95830c7a696 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 @@ -30,6 +30,7 @@ import org.testng.annotations.Test; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNull; @@ -56,6 +57,13 @@ public void testGetArgumentType() { assertEquals(FunctionUtils.getArgumentType((short) 1), PinotDataType.SHORT); } + @Test + public void testRequestTimeFunctionsAreNonDeterministic() { + assertFalse(FunctionRegistry.lookupFunctionInfo("now", 0).isDeterministic()); + assertFalse(FunctionRegistry.lookupFunctionInfo("ago", 1).isDeterministic()); + assertFalse(FunctionRegistry.lookupFunctionInfo("agomv", 1).isDeterministic()); + } + @Test public void testGetArgumentTypeForVendorTimestampSubclass() { // Vendor JDBC drivers commonly return Timestamp subclasses (e.g. BigQuery Simba's TimestampTz). 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 cfc2aea84949..6c443530b675 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 @@ * 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) { @@ -94,6 +88,22 @@ private PinotQuery compileToPinotQuery(String sql) { return query; } + private boolean containsFunction(Expression expression, String operator) { + if (expression == null || !expression.isSetFunctionCall()) { + return false; + } + Function function = expression.getFunctionCall(); + if (function.getOperator().equals(operator)) { + return true; + } + for (Expression operand : function.getOperands()) { + if (containsFunction(operand, operator)) { + return true; + } + } + return false; + } + @Test public void testCanonicalFunctionName() { Expression expression = compileToExpression("dIsTiNcT_cOuNt(AbC)"); @@ -241,8 +251,8 @@ public void testAggregationInCaseWhenStatements() { 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); + Assert.assertTrue(pinotQuery.getSelectList().get(0).isSetFunctionCall()); + Assert.assertTrue(containsFunction(pinotQuery.getSelectList().get(0), "now")); Assert.assertThrows(SqlCompilationException.class, () -> compileToPinotQuery("SELECT CASE WHEN 1 > 0 END FROM myTable")); @@ -2376,44 +2386,34 @@ public void testNoArgFunction() { @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); + Assert.assertTrue(containsFunction(pinotQuery.getFilterExpression(), "now")); 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(); - Assert.assertEquals(nowTs, 1577836800000L); + long timestamp = greaterThan.getOperands().get(1).getLiteral().getLongValue(); + Assert.assertEquals(timestamp, 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); + Assert.assertTrue(containsFunction(pinotQuery.getFilterExpression(), "ago")); query = "SELECT rand() FROM foo"; pinotQuery = compileToPinotQuery(query); @@ -2646,73 +2646,53 @@ public void testCompilationInvokedFunction() { public void testCompilationInvokedNestedFunctions() { 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); + Assert.assertTrue(containsFunction(pinotQuery.getFilterExpression(), "todatetime")); + Assert.assertTrue(containsFunction(pinotQuery.getFilterExpression(), "now")); } @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.assertTrue(containsFunction(expression, "now")); + 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.assertTrue(containsFunction(expression, "now")); + 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.assertTrue(containsFunction(expression, "now")); + 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.assertTrue(containsFunction(expression, "now")); + 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.assertTrue(containsFunction(expression, "now")); + 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.assertTrue(containsFunction(expression, "ago")); + 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.assertTrue(containsFunction(expression, "ago")); + Assert.assertFalse(expression.isSetLiteral()); expression = compileToExpression("toDateTime(millisSinceEpoch)"); Assert.assertNotNull(expression.getFunctionCall()); diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotDdlRestletResource.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotDdlRestletResource.java index 9a7c72caadd9..793ddf5af27b 100644 --- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotDdlRestletResource.java +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotDdlRestletResource.java @@ -600,7 +600,7 @@ private static boolean defaultValuesEqual(FieldSpec.DataType dataType, private void validateTableConfig(Schema schema, TableConfig tableConfig) { try { TableConfigValidationUtils.validateTableConfig(tableConfig, schema, null, - _pinotHelixResourceManager, _controllerConf, _pinotTaskManager); + _pinotHelixResourceManager, _controllerConf, _pinotTaskManager, null); } catch (ControllerApplicationException e) { throw e; } catch (IllegalArgumentException | IllegalStateException 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 d723fae45f5d..5ef9556264e3 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 @@ -254,7 +254,7 @@ public ConfigSuccessResponse addTable(String tableConfigStr, TableConfigTunerUtils.applyTunerConfigs(_pinotHelixResourceManager, tableConfig, schema, Map.of()); TableConfigValidationUtils.validateTableConfig( - tableConfig, schema, typesToSkip, _pinotHelixResourceManager, _controllerConf, _pinotTaskManager); + tableConfig, schema, typesToSkip, _pinotHelixResourceManager, _controllerConf, _pinotTaskManager, null); } catch (TableAlreadyExistsException e) { throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.CONFLICT, e); } catch (Exception e) { @@ -361,7 +361,7 @@ public CopyTableResponse copyTable( _pinotHelixResourceManager.addSchema(schema, true, false); LOGGER.info("[copyTable] Successfully added schema for table: {}", tableName); TableConfigValidationUtils.validateTableConfig( - realtimeTableConfig, schema, null, _pinotHelixResourceManager, _controllerConf, _pinotTaskManager); + realtimeTableConfig, schema, null, _pinotHelixResourceManager, _controllerConf, _pinotTaskManager, null); // 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); @@ -795,7 +795,8 @@ public ConfigSuccessResponse updateTableConfig( 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); 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 ec51fbccec7d..2ddf2d9681b6 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 @@ -46,21 +46,12 @@ public final class TableConfigValidationUtils { private TableConfigValidationUtils() { } - /** - * Validates a table config against the given schema and controller configuration. - * - * @param tableConfig the table config to validate - * @param schema the schema for the table (must not be null) - * @param typesToSkip comma-separated list of validation types to skip (ALL|TASK|UPSERT), or null - * @param resourceManager the Helix resource manager - * @param controllerConf the controller configuration - * @param taskManager the task manager, or null to skip task validation - */ public static void validateTableConfig(TableConfig tableConfig, Schema schema, @Nullable String typesToSkip, PinotHelixResourceManager resourceManager, - ControllerConf controllerConf, @Nullable PinotTaskManager taskManager) { + 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 8a0f60ddf48c..c58d09d2e7a0 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 @@ -591,7 +591,8 @@ private void validateConfig(TableConfigs tableConfigs, String database, @Nullabl 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); } @@ -601,7 +602,8 @@ private void validateConfig(TableConfigs tableConfigs, String database, @Nullabl 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/resources/PinotDdlRestletResourceMaterializedViewUnitTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotDdlRestletResourceMaterializedViewUnitTest.java index ef3755657da2..f21a874aa8bd 100644 --- a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotDdlRestletResourceMaterializedViewUnitTest.java +++ b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotDdlRestletResourceMaterializedViewUnitTest.java @@ -1064,7 +1064,7 @@ private QuietValidationMocks() { try { validation = Mockito.mockStatic(TableConfigValidationUtils.class); validation.when(() -> TableConfigValidationUtils.validateTableConfig( - any(), any(), any(), any(), any(), any())).then(invocation -> null); + any(), any(), any(), any(), any(), any(), any())).then(invocation -> null); tableResource = Mockito.mockStatic(PinotTableRestletResource.class); tableResource.when(() -> PinotTableRestletResource.tableTasksValidation(any(), any())) 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 68556333a24c..356864cddc43 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 @@ protected List getFieldConfigs() { @Override protected IngestionConfig getIngestionConfig() { List 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/validate/PinotTypeCoercionTest.java b/pinot-query-planner/src/test/java/org/apache/pinot/query/validate/PinotTypeCoercionTest.java index 6a771e266f31..0906ba4ab869 100644 --- a/pinot-query-planner/src/test/java/org/apache/pinot/query/validate/PinotTypeCoercionTest.java +++ b/pinot-query-planner/src/test/java/org/apache/pinot/query/validate/PinotTypeCoercionTest.java @@ -107,39 +107,29 @@ public void testBigintColumnVsTimestampLiteralKeepsColumnUnwrapped() { "Expected TIMESTAMP literal folded to BIGINT and column unwrapped. Got:\n" + plan); } - /** - * When a TIMESTAMP column is compared to a non-column TIMESTAMP-typed expression (e.g. {@code NOW() - 1000}, which - * after binary-arithmetic coercion is BIGINT-typed and then folded back to TIMESTAMP for the comparison), the - * resulting plan must keep the column unwrapped and constant-fold the right-hand side to a TIMESTAMP literal. - */ + /// When a TIMESTAMP column is compared to a non-column TIMESTAMP-typed expression (e.g. `NOW() - 1000`, + /// which after binary-arithmetic coercion is BIGINT-typed and then cast back to TIMESTAMP for the comparison), + /// the resulting plan must keep the column unwrapped. @Test - public void testTimestampColumnVsConstantSubexpressionKeepsColumnUnwrapped() { - // NOW() - 1000 is a constant for the lifetime of the query: arithmetic coercion casts NOW() to BIGINT, the - // subtraction is BIGINT - INT = BIGINT, and our comparison rule then casts that BIGINT result to TIMESTAMP. - // After constant folding, the right-hand side is rendered as a single TIMESTAMP literal. + public void testTimestampColumnVsNonDeterministicSubexpressionKeepsColumnUnwrapped() { String plan = explain("SELECT ts_timestamp FROM a WHERE ts_timestamp > NOW() - 1000"); assertFalse(plan.contains("CAST(" + TS_TIMESTAMP_ORD + ")"), - "TIMESTAMP column should not be wrapped in CAST when right-hand side is constant. Got:\n" + plan); - // The fractional part is optional here: NOW()/ago() produce non-deterministic millis that may land on a whole - // second (rendered without a fractional part). The deterministic millis-preservation guard is in - // testTimestampColumnVsSubSecondLiteralPreservesMillis below. + "TIMESTAMP column should not be wrapped in CAST when right-hand side is non-column expression. Got:\n" + plan); assertTrue( - plan.matches("(?s).*>\\(\\" + TS_TIMESTAMP_ORD + ", \\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}(\\.\\d+)?\\).*"), - "Right-hand side should be constant-folded to a TIMESTAMP literal. Got:\n" + plan); + plan.contains(">(" + TS_TIMESTAMP_ORD + ", CAST(-(CAST(NOW()):BIGINT NOT NULL, 1000)):TIMESTAMP(3)"), + "Right-hand side should preserve the non-deterministic NOW() call. Got:\n" + plan); } - /** - * When a BIGINT column is compared to a non-column TIMESTAMP-typed expression (e.g. {@code NOW()}), the existing - * behavior of casting the TIMESTAMP side to BIGINT must be preserved: the BIGINT column stays unwrapped, and the - * TIMESTAMP function is folded to a BIGINT literal. - */ + /// When a BIGINT column is compared to a non-column TIMESTAMP-typed expression (e.g. `NOW()`), the existing + /// behavior of casting the TIMESTAMP side to BIGINT must be preserved: the BIGINT column stays unwrapped, and the + /// non-deterministic TIMESTAMP function is not constant-folded. @Test public void testBigintColumnVsTimestampFunctionKeepsColumnUnwrapped() { String plan = explain("SELECT col7 FROM a WHERE col7 < NOW()"); assertFalse(plan.contains("CAST(" + COL7_ORD + ")"), "BIGINT column should not be wrapped in CAST. Got:\n" + plan); - assertTrue(plan.matches("(?s).*<\\(\\" + COL7_ORD + ", \\d+\\).*"), - "Right-hand side should be folded to a BIGINT literal. Got:\n" + plan); + assertTrue(plan.contains("<(" + COL7_ORD + ", CAST(NOW()):BIGINT"), + "Right-hand side should preserve the non-deterministic NOW() call. Got:\n" + plan); } /** @@ -166,24 +156,18 @@ public void testTimestampColumnVsTimestampFunctionHasNoCast() { "No CAST should appear in the plan. Got:\n" + plan); } - /** - * Regression for the {@code ago()}-style use case: {@code __time > ago('PT5M')} should keep the column unwrapped. - * {@code ago(String)} is a scalar function that returns {@code long} (rendered as BIGINT in SQL), so the comparison - * is TIMESTAMP-vs-BIGINT and the new rule applies. Before this rule, the column was wrapped in {@code CAST(.. AS - * BIGINT)} per row, which made the query significantly slower than the workaround {@code __time > concat(ago(...), - * '')} that happened to push the cast onto the literal side via the VARCHAR coercion path. - */ + /// Regression for the `ago()`-style use case: `__time > ago('PT5M')` should keep the column unwrapped. + /// `ago(String)` is a scalar function that returns `long` (rendered as BIGINT in SQL), so the comparison is + /// TIMESTAMP-vs-BIGINT and the new rule applies. Before this rule, the column was wrapped in `CAST(.. AS BIGINT)` + /// per row, which made the query significantly slower than the workaround `__time > concat(ago(...), '')` that + /// happened to push the cast onto the literal side via the VARCHAR coercion path. @Test public void testTimestampColumnVsAgoFunctionKeepsColumnUnwrapped() { String plan = explain("SELECT ts_timestamp FROM a WHERE ts_timestamp > ago('PT5M')"); assertFalse(plan.contains("CAST(" + TS_TIMESTAMP_ORD + ")"), "TIMESTAMP column should not be wrapped in CAST when compared to ago(...). Got:\n" + plan); - // The fractional part is optional here: NOW()/ago() produce non-deterministic millis that may land on a whole - // second (rendered without a fractional part). The deterministic millis-preservation guard is in - // testTimestampColumnVsSubSecondLiteralPreservesMillis below. - assertTrue( - plan.matches("(?s).*>\\(\\" + TS_TIMESTAMP_ORD + ", \\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}(\\.\\d+)?\\).*"), - "Right-hand side should be constant-folded to a TIMESTAMP literal. Got:\n" + plan); + assertTrue(plan.contains(">(" + TS_TIMESTAMP_ORD + ", CAST(AGO(_UTF-8'PT5M')):TIMESTAMP(3)"), + "Right-hand side should preserve the non-deterministic ago(...) call. Got:\n" + plan); } /** 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 5dd51a79e317..ca19efe9042e 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.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; @@ -172,15 +174,20 @@ public static void validate(TableConfig tableConfig, Schema schema) { * 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 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 +195,7 @@ public static void validate(TableConfig tableConfig, Schema schema, @Nullable St } private static void validateEffectiveTableConfig(TableConfig tableConfig, Schema schema, - Set skipTypes) { + Set skipTypes, @Nullable TableConfig existingTableConfig) { // Sanitize the table config before validation sanitize(tableConfig); @@ -198,7 +205,7 @@ private static void validateEffectiveTableConfig(TableConfig tableConfig, Schema validateValidationConfig(tableConfig, schema); validateSegmentAssignmentConfig(tableConfig); - validateIngestionConfig(tableConfig, schema); + validateIngestionConfig(tableConfig, schema, existingTableConfig); if (tableConfig.getTableType() == TableType.REALTIME) { validateStreamConfigMaps(tableConfig); } @@ -209,7 +216,7 @@ private static void validateEffectiveTableConfig(TableConfig tableConfig, Schema validateInstanceAssignmentConfigs(tableConfig); if (!skipTypes.contains(ValidationType.UPSERT)) { - validateUpsertAndDedupConfig(tableConfig, schema); + validateUpsertAndDedupConfig(tableConfig, schema, existingTableConfig); } validateTaskConfig(tableConfig); @@ -458,7 +465,8 @@ private static boolean isValidPeerDownloadScheme(String peerSegmentDownloadSchem /// - Complex-type config: no schema field collides with a `prefixesToRename` prefix. /// - Schema-conforming transformer config. @VisibleForTesting - public static void validateIngestionConfig(TableConfig tableConfig, Schema schema) { + 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 aggregationSourceColumns = validateMetricsAggregation(tableConfig, schema); @@ -566,6 +574,7 @@ public static void validateIngestionConfig(TableConfig tableConfig, Schema schem // Transform configs List transformConfigs = ingestionConfig.getTransformConfigs(); if (transformConfigs != null) { + List 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 +620,7 @@ public static void validateIngestionConfig(TableConfig tableConfig, Schema schem + columnName + "'"); } try { + validateDeterministicTransformFunction(transformConfig, existingTransformConfigs); expressionEvaluator = FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction); } catch (Exception e) { throw new IllegalStateException( @@ -649,6 +659,63 @@ public static void validateIngestionConfig(TableConfig tableConfig, Schema schem } } + @Nullable + private static List getTransformConfigs(@Nullable TableConfig tableConfig) { + IngestionConfig ingestionConfig = tableConfig != null ? tableConfig.getIngestionConfig() : null; + return ingestionConfig != null ? ingestionConfig.getTransformConfigs() : null; + } + + @Nullable + private static List getPostPartialUpsertTransformConfigs(@Nullable TableConfig tableConfig) { + UpsertConfig upsertConfig = tableConfig != null ? tableConfig.getUpsertConfig() : null; + return upsertConfig != null ? upsertConfig.getPostPartialUpsertTransformConfigs() : null; + } + + private static void validateDeterministicTransformFunction(TransformConfig transformConfig, + @Nullable List existingTransformConfigs) { + if (hasSameTransformConfig(existingTransformConfigs, transformConfig)) { + return; + } + String transformFunction = transformConfig.getTransformFunction(); + if (FunctionEvaluatorFactory.isGroovyExpression(transformFunction)) { + return; + } + validateDeterministicExpression(RequestContextUtils.getExpression(transformFunction), transformFunction); + } + + private static boolean hasSameTransformConfig(@Nullable List 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 validateDeterministicExpression(ExpressionContext expression, String transformFunction) { + if (expression.getType() != ExpressionContext.Type.FUNCTION) { + return; + } + FunctionContext function = expression.getFunction(); + List arguments = function.getArguments(); + for (ExpressionContext argument : arguments) { + validateDeterministicExpression(argument, transformFunction); + } + String functionName = function.getFunctionName(); + String canonicalName = FunctionRegistry.canonicalize(functionName); + FunctionInfo functionInfo = FunctionRegistry.lookupFunctionInfo(canonicalName, arguments.size()); + if (functionInfo != null && !functionInfo.isDeterministic()) { + throw new IllegalStateException( + String.format("Non-deterministic function '%s' is not allowed in transform function: %s", functionName, + 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 +979,11 @@ static void validateStreamConfig(StreamConfig streamConfig) { /// - 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 +1111,8 @@ static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema schema) List postPartialUpsertTransformConfigs = upsertConfig.getPostPartialUpsertTransformConfigs(); if (postPartialUpsertTransformConfigs != null) { + List existingPostPartialUpsertTransformConfigs = + getPostPartialUpsertTransformConfigs(existingTableConfig); Preconditions.checkState(upsertConfig.getMode() == UpsertConfig.Mode.PARTIAL, "postPartialUpsertTransformConfigs can only be configured for PARTIAL upsert tables"); Set primaryKeyColumns = new HashSet<>(schema.getPrimaryKeyColumns()); @@ -1072,6 +1146,7 @@ static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema schema) + columnName + "' in postPartialUpsertTransformConfigs"); } try { + validateDeterministicTransformFunction(transformConfig, existingPostPartialUpsertTransformConfigs); FunctionEvaluator expressionEvaluator = FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction); List 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 e465874aae18..593ec59d1901 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 void testTransformReturningNullDoesNotOverrideExistingBytesValue() { 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 3bd121218121..95910763f84f 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 @@ -385,6 +385,32 @@ public void validateIngestionConfig() { 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("Non-deterministic function 'now'")); + + 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("Non-deterministic function 'now'")); + + ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol", "rand()"))); + nonDeterministicError = + expectThrows(IllegalStateException.class, () -> TableConfigUtils.validate(tableConfig, transformSchema)); + assertTrue(nonDeterministicError.getMessage().contains("Non-deterministic function 'rand'")); + + ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol", "rand(123)"))); + TableConfigUtils.validate(tableConfig, schema); + // valid transform configs schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME) .addSingleValueDimension("myCol", DataType.STRING) @@ -581,7 +607,7 @@ public void ingestionSourceFieldConfigsTest() { new SourceFieldConfig("shared", PinotDataType.LONG, true), new SourceFieldConfig("shared", PinotDataType.INT, false) )); - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); // The same field twice within the same phase is rejected. ingestionConfig.setSourceFieldConfigs(List.of( @@ -589,7 +615,7 @@ public void ingestionSourceFieldConfigsTest() { new SourceFieldConfig("dup", PinotDataType.INT, false) )); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail on duplicate SourceFieldConfig within the same phase"); } catch (IllegalStateException e) { // expected @@ -608,7 +634,7 @@ public void ingestionAggregationConfigsTest() { .setIngestionConfig(ingestionConfig) .build(); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail due to destination column not being in schema"); } catch (IllegalStateException e) { // expected @@ -617,7 +643,7 @@ public void ingestionAggregationConfigsTest() { schema.addField(new DimensionFieldSpec("d1", DataType.DOUBLE, true)); tableConfig.getIndexingConfig().setAggregateMetrics(true); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail due to aggregateMetrics being set"); } catch (IllegalStateException e) { // expected @@ -625,7 +651,7 @@ public void ingestionAggregationConfigsTest() { tableConfig.getIndexingConfig().setAggregateMetrics(false); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail due to aggregation column being a dimension"); } catch (IllegalStateException e) { // expected @@ -640,7 +666,7 @@ public void ingestionAggregationConfigsTest() { ingestionConfig.setAggregationConfigs(List.of(new AggregationConfig(null, null))); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail due to null columnName/aggregationFunction"); } catch (IllegalStateException e) { // expected @@ -649,7 +675,7 @@ public void ingestionAggregationConfigsTest() { ingestionConfig.setAggregationConfigs( Arrays.asList(new AggregationConfig("m1", "SUM(s1)"), new AggregationConfig("m1", "SUM(s2)"))); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail due to duplicate destination column"); } catch (IllegalStateException e) { // expected @@ -657,7 +683,7 @@ public void ingestionAggregationConfigsTest() { ingestionConfig.setAggregationConfigs(List.of(new AggregationConfig("m1", "SUM s1"))); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail due to invalid aggregation function"); } catch (IllegalStateException e) { // expected @@ -665,22 +691,22 @@ public void ingestionAggregationConfigsTest() { ingestionConfig.setAggregationConfigs(List.of(new AggregationConfig("m1", "SUM(s1 - s2)"))); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail due to inner value not being a column"); } catch (IllegalStateException e) { // expected } ingestionConfig.setAggregationConfigs(List.of(new AggregationConfig("m1", "SUM(m1)"))); - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); ingestionConfig.setAggregationConfigs(List.of(new AggregationConfig("m1", "SUM(s1)"))); - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); schema.addField(new MetricFieldSpec("m2", DataType.DOUBLE)); indexingConfig.setNoDictionaryColumns(List.of("m1", "m2")); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail due to one metric column not being aggregated"); } catch (IllegalStateException e) { // expected @@ -697,7 +723,7 @@ public void ingestionAggregationConfigsTest() { .build(); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail due to not supported aggregation function"); } catch (IllegalStateException e) { // expected @@ -724,7 +750,7 @@ public void ingestionAggregationConfigsTest() { .build(); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); } catch (IllegalStateException e) { fail("Should not fail due to valid aggregation function", e); } @@ -744,7 +770,7 @@ public void ingestionAggregationConfigsTest() { .build(); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); } catch (IllegalStateException e) { fail("Should not fail due to valid aggregation function", e); } @@ -761,7 +787,7 @@ public void ingestionAggregationConfigsTest() { .build(); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); } catch (IllegalStateException e) { fail("Log2m should have defaulted to 8", e); } @@ -775,7 +801,7 @@ public void ingestionAggregationConfigsTest() { .build(); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail due to multiple arguments"); } catch (IllegalArgumentException e) { // expected @@ -800,7 +826,7 @@ public void ingestionAggregationConfigsTest() { .setIngestionConfig(ingestionConfig) .setNoDictionaryColumns(List.of("d1", "d2", "d3", "d4", "d5")) .build(); - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); // with too many arguments should fail schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME) @@ -818,7 +844,7 @@ public void ingestionAggregationConfigsTest() { .build(); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should have failed with too many arguments but didn't"); } catch (IllegalStateException e) { // Expected @@ -909,13 +935,13 @@ public void metricsAggregationValidationTest() { .setAggregateMetrics(true) .setNoDictionaryColumns(List.of("m1")) .build(); - TableConfigUtils.validateIngestionConfig(validConfig, schemaWithSvColumns); + TableConfigUtils.validateIngestionConfig(validConfig, schemaWithSvColumns, null); } private void assertMetricsAggregationValidationFails(TableConfig tableConfig, Schema schema, String expectedMessageSubstring) { try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail with message containing: " + expectedMessageSubstring); } catch (IllegalStateException e) { assertTrue(e.getMessage().contains(expectedMessageSubstring), @@ -938,14 +964,14 @@ public void ingestionStreamConfigsTest() { // Multiple stream configs are allowed try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); } catch (IllegalStateException e) { fail("Multiple stream configs should be supported"); } // stream config should be valid ingestionConfig.setStreamIngestionConfig(new StreamIngestionConfig(List.of(streamConfigs))); - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); // validate the proto decoder streamConfigs = getKafkaStreamConfigs(); @@ -1084,19 +1110,19 @@ public void ingestionServerIngestionOomProtectionTest() { TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME) .setIngestionConfig(ingestionConfig) .build(); - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); streamIngestionConfig.setOomProtection(Enablement.ENABLE); - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); streamIngestionConfig.setOomProtection(Enablement.DISABLE); - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); streamIngestionConfig.setOomProtection(Enablement.DEFAULT); - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); streamIngestionConfig.setOomProtection(null); - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); } @Test @@ -1117,7 +1143,7 @@ public void ingestionBatchConfigsTest() { new BatchIngestionConfig(Arrays.asList(batchConfigMap, batchConfigMap), null, null)); TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(ingestionConfig).build(); - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); } @Test @@ -1140,12 +1166,12 @@ public void ingestionConfigForDimensionTableTest() { .setIsDimTable(true) .setIngestionConfig(ingestionConfig) .build(); - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); // dimension tables should have batch ingestion config ingestionConfig.setBatchIngestionConfig(null); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail for Dimension table without batch ingestion config"); } catch (IllegalStateException e) { // expected @@ -1155,7 +1181,7 @@ public void ingestionConfigForDimensionTableTest() { ingestionConfig.setBatchIngestionConfig( new BatchIngestionConfig(List.of(batchConfigMap), "APPEND", null)); try { - TableConfigUtils.validateIngestionConfig(tableConfig, schema); + TableConfigUtils.validateIngestionConfig(tableConfig, schema, null); fail("Should fail for Dimension table with ingestion type APPEND (should be REFRESH)"); } catch (IllegalStateException e) { // expected @@ -2634,7 +2660,7 @@ public void testValidateUpsertConfig() { .setAggregateMetrics(true) .build(); try { - TableConfigUtils.validateIngestionConfig(tableConfig, validSchema); + TableConfigUtils.validateIngestionConfig(tableConfig, validSchema, null); fail(); } catch (IllegalStateException e) { assertEquals(e.getMessage(), "Metrics aggregation and upsert cannot be enabled together"); @@ -2649,7 +2675,7 @@ public void testValidateUpsertConfig() { .setIngestionConfig(ingestionConfig) .build(); try { - TableConfigUtils.validateIngestionConfig(tableConfig, validSchema); + TableConfigUtils.validateIngestionConfig(tableConfig, validSchema, null); fail(); } catch (IllegalStateException e) { assertEquals(e.getMessage(), "Metrics aggregation and upsert cannot be enabled together"); @@ -2663,7 +2689,7 @@ public void testValidateUpsertConfig() { .setIngestionConfig(ingestionConfig) .build(); try { - TableConfigUtils.validateIngestionConfig(tableConfig, validSchema); + TableConfigUtils.validateIngestionConfig(tableConfig, validSchema, null); fail(); } catch (IllegalStateException e) { assertEquals(e.getMessage(), 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 2fe1b6a07517..22d718fae0f7 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 dd84b3e17526..3fe48d17bbb6 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 58e0967c8566..2b2e2a47d64f 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" } ] },