This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 2aa3aad815d Allow non-immutable ingestion transforms for offline
tables (#19244)
2aa3aad815d is described below
commit 2aa3aad815d58114bb1761f4ab14fb1224830e0d
Author: Xiang Fu <[email protected]>
AuthorDate: Thu Aug 13 14:26:55 2026 -0700
Allow non-immutable ingestion transforms for offline tables (#19244)
---
.../api/PinotTableRestletResourceTest.java | 13 ++++-----
.../api/TableConfigsRestletResourceTest.java | 20 ++++++-------
.../apache/pinot/core/util/SchemaUtilsTest.java | 13 ++++++---
.../segment/local/utils/TableConfigUtils.java | 7 +++--
.../segment/local/utils/TableConfigUtilsTest.java | 33 ++++++++++++----------
5 files changed, 48 insertions(+), 38 deletions(-)
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTableRestletResourceTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTableRestletResourceTest.java
index a6c9f8f1a36..4a4dfc54ce0 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTableRestletResourceTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTableRestletResourceTest.java
@@ -535,10 +535,9 @@ public class PinotTableRestletResourceTest extends
ControllerTest {
throws Exception {
String tableName = "legacyNonDeterministicTransform";
DEFAULT_INSTANCE.addDummySchema(tableName);
-
IngestionConfig ingestionConfig = new IngestionConfig();
- ingestionConfig.setTransformConfigs(List.of(new TransformConfig("dimA",
"now()")));
- TableConfig legacyTableConfig = getOfflineTableBuilder(tableName)
+ ingestionConfig.setTransformConfigs(List.of(new
TransformConfig("timeColumn", "now()")));
+ TableConfig legacyTableConfig = getRealtimeTableBuilder(tableName)
.setIngestionConfig(ingestionConfig)
.build();
@@ -550,20 +549,20 @@ public class PinotTableRestletResourceTest extends
ControllerTest {
// Seed the config below the REST validation layer to model a table
persisted before this validation existed.
DEFAULT_INSTANCE.getHelixResourceManager().addTable(legacyTableConfig);
- TableConfig update = getTableConfig(tableName, "OFFLINE");
+ TableConfig update = getTableConfig(tableName, "REALTIME");
update.getValidationConfig().setRetentionTimeValue("10");
JsonNode validationResponse =
JsonUtils.stringToJsonNode(tableClient().validateTableConfig(update.toJsonString()));
- assertTrue(validationResponse.has("OFFLINE"));
+ assertTrue(validationResponse.has("REALTIME"));
JsonNode response = JsonUtils.stringToJsonNode(updateTable(tableName,
update.toJsonString()));
assertTrue(response.has("status"));
- TableConfig stored = getTableConfig(tableName, "OFFLINE");
+ TableConfig stored = getTableConfig(tableName, "REALTIME");
assertEquals(stored.getValidationConfig().getRetentionTimeValue(), "10");
assertEquals(stored.getIngestionConfig().getTransformConfigs().get(0).getTransformFunction(),
"now()");
IngestionConfig changedIngestionConfig = new IngestionConfig();
- changedIngestionConfig.setTransformConfigs(List.of(new
TransformConfig("dimA", "plus(now(), 1)")));
+ changedIngestionConfig.setTransformConfigs(List.of(new
TransformConfig("timeColumn", "plus(now(), 1)")));
update.setIngestionConfig(changedIngestionConfig);
PinotAdminException validationError =
expectThrows(PinotAdminException.class, () ->
tableClient().validateTableConfig(update.toJsonString()));
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
index e7bbccbbcac..f3b332271c1 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
@@ -519,12 +519,12 @@ public class TableConfigsRestletResourceTest extends
ControllerTest {
Schema schema = createDummySchema(tableName);
IngestionConfig ingestionConfig = new IngestionConfig();
- ingestionConfig.setTransformConfigs(List.of(new TransformConfig("dimA",
"now()")));
- TableConfig legacyOfflineConfig = getBaseTableConfigBuilder(tableName,
TableType.OFFLINE)
+ ingestionConfig.setTransformConfigs(List.of(new
TransformConfig("timeColumn", "now()")));
+ TableConfig legacyRealtimeConfig = getBaseTableConfigBuilder(tableName,
TableType.REALTIME)
.setIngestionConfig(ingestionConfig)
.build();
- TableConfigs tableConfigs = new TableConfigs(tableName, schema,
legacyOfflineConfig, null);
- String tableNameWithType =
TableNameBuilder.OFFLINE.tableNameWithType(tableName);
+ TableConfigs tableConfigs = new TableConfigs(tableName, schema, null,
legacyRealtimeConfig);
+ String tableNameWithType =
TableNameBuilder.REALTIME.tableNameWithType(tableName);
try {
String createError = Assert.expectThrows(Exception.class,
() ->
adminClient.getTableClient().createTableConfigs(tableConfigs.toPrettyJsonString(),
null, null))
@@ -533,21 +533,21 @@ public class TableConfigsRestletResourceTest extends
ControllerTest {
// Seed the config below the REST validation layer to model a table
persisted before this validation existed.
DEFAULT_INSTANCE.addSchema(schema);
- DEFAULT_INSTANCE.getHelixResourceManager().addTable(legacyOfflineConfig);
+
DEFAULT_INSTANCE.getHelixResourceManager().addTable(legacyRealtimeConfig);
TableConfigs update =
adminClient.getTableClient().getTableConfigsObject(tableName);
- update.getOffline().getValidationConfig().setRetentionTimeValue("10");
+ update.getRealtime().getValidationConfig().setRetentionTimeValue("10");
adminClient.getTableClient()
.updateTableConfigs(tableName, update.toPrettyJsonString(), null,
false, false);
TableConfigs stored =
adminClient.getTableClient().getTableConfigsObject(tableName);
-
Assert.assertEquals(stored.getOffline().getValidationConfig().getRetentionTimeValue(),
"10");
-
Assert.assertEquals(stored.getOffline().getIngestionConfig().getTransformConfigs().get(0).getTransformFunction(),
+
Assert.assertEquals(stored.getRealtime().getValidationConfig().getRetentionTimeValue(),
"10");
+
Assert.assertEquals(stored.getRealtime().getIngestionConfig().getTransformConfigs().get(0).getTransformFunction(),
"now()");
IngestionConfig changedIngestionConfig = new IngestionConfig();
- changedIngestionConfig.setTransformConfigs(List.of(new
TransformConfig("dimA", "plus(now(), 1)")));
- update.getOffline().setIngestionConfig(changedIngestionConfig);
+ changedIngestionConfig.setTransformConfigs(List.of(new
TransformConfig("timeColumn", "plus(now(), 1)")));
+ update.getRealtime().setIngestionConfig(changedIngestionConfig);
String updateError = Assert.expectThrows(Exception.class,
() -> adminClient.getTableClient()
.updateTableConfigs(tableName, update.toPrettyJsonString(),
null, false, false))
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/util/SchemaUtilsTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/util/SchemaUtilsTest.java
index 336f984ae5d..ad87a56909e 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/util/SchemaUtilsTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/util/SchemaUtilsTest.java
@@ -190,12 +190,17 @@ public class SchemaUtilsTest {
@Test
public void testCompatibilityGrandfathersExistingNonDeterministicTransform()
{
- Schema schema =
- new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addMetric("eventTimeMs",
DataType.LONG).build();
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addMetric("eventTimeMs", DataType.LONG)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .build();
IngestionConfig ingestionConfig = new IngestionConfig();
ingestionConfig.setTransformConfigs(List.of(new
TransformConfig("eventTimeMs", "now()")));
- TableConfig tableConfig =
- new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(ingestionConfig).build();
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName(TIME_COLUMN)
+ .setStreamConfigs(getStreamConfigs())
+ .setIngestionConfig(ingestionConfig)
+ .build();
// A new table using this transform is still rejected.
Assert.expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validate(tableConfig, schema));
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 fc2dcd1e84b..8ef3c48727e 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
@@ -457,7 +457,8 @@ public final class TableConfigUtils {
/// - Enrichment configs: each config is valid.
/// - Transform configs: non-null column and function, no duplicate
destination, and each destination is a schema
/// column, an intermediate consumed by another transform, or an
aggregation source column; the function is valid
- /// and does not reference its own destination.
+ /// and does not reference its own destination. REALTIME tables require
immutable functions, while OFFLINE tables
+ /// allow all volatility categories.
/// - Complex-type config: no schema field collides with a
`prefixesToRename` prefix.
/// - Schema-conforming transformer config.
@VisibleForTesting
@@ -621,7 +622,9 @@ public final class TableConfigUtils {
+ columnName + "'");
}
try {
- validateIngestionTransformFunctionVolatility(transformConfig,
existingTransformConfigs);
+ if (tableConfig.getTableType() == TableType.REALTIME) {
+ validateIngestionTransformFunctionVolatility(transformConfig,
existingTransformConfigs);
+ }
expressionEvaluator =
FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction);
} catch (Exception e) {
throw new IllegalStateException(
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 af4d7a85875..29ed034bcfa 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
@@ -381,7 +381,9 @@ public class TableConfigUtilsTest {
TableConfigUtils.validate(tableConfig, schema);
// valid transform configs
- schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol",
DataType.STRING)
+ schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
.build();
indexingConfig.setNoDictionaryColumns(List.of("myCol"));
ingestionConfig.setAggregationConfigs(null);
@@ -390,35 +392,36 @@ public class TableConfigUtilsTest {
Schema transformSchema = schema;
ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol",
"now()")));
- IllegalStateException nonDeterministicError =
- expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validate(tableConfig, transformSchema));
- assertTrue(nonDeterministicError.getMessage().contains("Function 'now' has
VOLATILE volatility"));
+ TableConfigUtils.validate(tableConfig, transformSchema);
- IngestionConfig existingIngestionConfig = new IngestionConfig();
- existingIngestionConfig.setTransformConfigs(List.of(new
TransformConfig("myCol", "now()")));
- TableConfig existingTableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setIngestionConfig(existingIngestionConfig)
+ TableConfig realtimeTableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName(TIME_COLUMN)
+ .setStreamConfigs(getStreamConfigs())
+ .setIngestionConfig(ingestionConfig)
.build();
- TableConfigUtils.validate(tableConfig, schema, null, existingTableConfig);
+ IllegalStateException nonDeterministicError =
expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validate(realtimeTableConfig, transformSchema));
+ assertTrue(nonDeterministicError.getMessage().contains("Function 'now' has
VOLATILE volatility"));
+ TableConfigUtils.validate(realtimeTableConfig, transformSchema, null,
realtimeTableConfig);
ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol",
"plus(now(), 1)")));
nonDeterministicError = expectThrows(IllegalStateException.class,
- () -> TableConfigUtils.validate(tableConfig, transformSchema, null,
existingTableConfig));
+ () -> TableConfigUtils.validate(realtimeTableConfig, transformSchema));
assertTrue(nonDeterministicError.getMessage().contains("Function 'now' has
VOLATILE volatility"));
ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol",
"rand()")));
- nonDeterministicError =
- expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validate(tableConfig, transformSchema));
+ nonDeterministicError = expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validate(realtimeTableConfig, transformSchema));
assertTrue(nonDeterministicError.getMessage().contains("Function 'rand'
has VOLATILE volatility"));
ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol",
"reqId('unused')")));
- nonDeterministicError =
- expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validate(tableConfig, transformSchema));
+ nonDeterministicError = expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validate(realtimeTableConfig, transformSchema));
assertTrue(nonDeterministicError.getMessage().contains("Function 'reqid'
has STABLE volatility"),
nonDeterministicError.getMessage());
ingestionConfig.setTransformConfigs(List.of(new TransformConfig("myCol",
"rand(123)")));
- TableConfigUtils.validate(tableConfig, schema);
+ TableConfigUtils.validate(realtimeTableConfig, transformSchema);
// Legacy schema-level transforms are also part of the ingestion pipeline.
A new table must reject them, while
// validation of an existing table stays permissive so unrelated config
updates are not stranded.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
