You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "deepthi912 (via GitHub)" <gi...@apache.org> on 2026/02/05 21:01:34 UTC
[PR] Add checks to not update upsert configs after table creation [pinot]
deepthi912 opened a new pull request, #17645:
URL: https://github.com/apache/pinot/pull/17645
**Reference:** Recently we have seen users changing comparison columns, primary keys which is not a recommended operation for an upsert table. Changes in such configs can lead to data inconsistencies when server restarts and commits segments during this time. This hard check ensures that user doesn't make such changes once the table is created, best way to move forward with such changes is to recreate the table to avoid any discrepencies in the data across the replicas.
Documentation: https://docs.pinot.apache.org/manage-data/data-import/upsert-and-dedup/upsert#faq
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add checks to not update upsert / dedup configs after table creation [pinot]
Posted by "deepthi912 (via GitHub)" <gi...@apache.org>.
deepthi912 commented on code in PR #17645:
URL: https://github.com/apache/pinot/pull/17645#discussion_r2785282213
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -1173,6 +1174,156 @@ static void validatePartialUpsertStrategies(TableConfig tableConfig, Schema sche
}
}
+ /**
+ * Validates that critical upsert configuration fields are not changed during table config update.
+ * Checks: mode, hashFunction, comparisonColumns, timeColumn (when no comparison columns),
+ * deleteRecordColumn, dropOutOfOrderRecord, outOfOrderRecordColumn,
+ * partialUpsertStrategies, defaultPartialUpsertStrategy.
+ *
+ * @param existingConfig the existing table config
+ * @param newConfig the new table config being applied
+ * @param force if true, logs a warning instead of throwing an exception
+ * @throws IllegalArgumentException if any critical config field is changed and force is false
+ */
+ public static void validateUpsertConfigUpdate(TableConfig existingConfig, TableConfig newConfig, boolean force) {
+ String tableName = existingConfig.getTableName();
+ List<String> violations = new ArrayList<>();
+
+ boolean existingUpsertEnabled = existingConfig.isUpsertEnabled();
+ boolean newUpsertEnabled = newConfig.isUpsertEnabled();
+
+ // Check if upsert is being added or removed
+ if (existingUpsertEnabled != newUpsertEnabled) {
+ if (existingUpsertEnabled) {
+ violations.add("upsertConfig cannot remove from existing upsert table");
+ } else {
+ violations.add("upsertConfig cannot add to existing non-upsert table");
+ }
+ } else if (existingUpsertEnabled) {
+ UpsertConfig existingUpsertConfig = existingConfig.getUpsertConfig();
+ UpsertConfig newUpsertConfig = newConfig.getUpsertConfig();
+
+ if (existingUpsertConfig.getMode() != newUpsertConfig.getMode()) {
+ violations.add(
+ String.format("mode (%s -> %s)", existingUpsertConfig.getMode(), newUpsertConfig.getMode()));
+ }
+ if (existingUpsertConfig.getHashFunction() != newUpsertConfig.getHashFunction()) {
+ violations.add(String.format("hashFunction (%s -> %s)", existingUpsertConfig.getHashFunction(),
+ newUpsertConfig.getHashFunction()));
+ }
+ if (!Objects.equals(existingUpsertConfig.getComparisonColumns(),
+ newUpsertConfig.getComparisonColumns())) {
+ violations.add(
+ String.format("comparisonColumns (%s -> %s)", existingUpsertConfig.getComparisonColumns(),
+ newUpsertConfig.getComparisonColumns()));
+ }
+ List<String> existingComparisonColumns = existingUpsertConfig.getComparisonColumns();
+ if (existingComparisonColumns == null || existingComparisonColumns.isEmpty()) {
+ String existingTimeColumn =
+ existingConfig.getValidationConfig() != null ? existingConfig.getValidationConfig().getTimeColumnName()
+ : null;
+ String newTimeColumn =
+ newConfig.getValidationConfig() != null ? newConfig.getValidationConfig().getTimeColumnName() : null;
+ if (!Objects.equals(existingTimeColumn, newTimeColumn)) {
+ violations.add(
+ String.format("timeColumnName (%s -> %s) - used as default comparison column", existingTimeColumn,
+ newTimeColumn));
+ }
+ }
+ if (existingUpsertConfig.isDropOutOfOrderRecord() != newUpsertConfig.isDropOutOfOrderRecord()) {
+ violations.add(
+ String.format("dropOutOfOrderRecord (%s -> %s)", existingUpsertConfig.isDropOutOfOrderRecord(),
+ newUpsertConfig.isDropOutOfOrderRecord()));
+ }
+ if (!Objects.equals(existingUpsertConfig.getOutOfOrderRecordColumn(),
+ newUpsertConfig.getOutOfOrderRecordColumn())) {
+ violations.add(String.format("outOfOrderRecordColumn (%s -> %s)",
+ existingUpsertConfig.getOutOfOrderRecordColumn(), newUpsertConfig.getOutOfOrderRecordColumn()));
+ }
+ if (existingUpsertConfig.getMode() == UpsertConfig.Mode.PARTIAL) {
+ if (!Objects.equals(existingUpsertConfig.getPartialUpsertStrategies(),
+ newUpsertConfig.getPartialUpsertStrategies())) {
+ violations.add(String.format("partialUpsertStrategies (%s -> %s)",
+ existingUpsertConfig.getPartialUpsertStrategies(), newUpsertConfig.getPartialUpsertStrategies()));
+ }
+ if (existingUpsertConfig.getDefaultPartialUpsertStrategy()
+ != newUpsertConfig.getDefaultPartialUpsertStrategy()) {
+ violations.add(String.format("defaultPartialUpsertStrategy (%s -> %s)",
+ existingUpsertConfig.getDefaultPartialUpsertStrategy(),
+ newUpsertConfig.getDefaultPartialUpsertStrategy()));
+ }
+ }
+ }
+ handleViolations(tableName, violations, force, "upsert");
+ }
+
+ /**
+ * Validates that critical dedup configuration fields are not changed during table config update.
+ * Checks: dedupEnabled, hashFunction, dedupTimeColumn, timeColumnName (when dedupTimeColumn not specified).
+ *
+ * @param existingConfig the existing table config
+ * @param newConfig the new table config being applied
+ * @param force if true, logs a warning instead of throwing an exception
+ * @throws IllegalArgumentException if any critical config field is changed and force is false
+ */
+ public static void validateDedupConfigUpdate(TableConfig existingConfig, TableConfig newConfig, boolean force) {
+ String tableName = existingConfig.getTableName();
+ List<String> violations = new ArrayList<>();
+
+ boolean existingDedupEnabled = existingConfig.isDedupEnabled();
+ boolean newDedupEnabled = newConfig.isDedupEnabled();
+ if (existingDedupEnabled != newDedupEnabled) {
+ if (existingDedupEnabled) {
+ violations.add("dedupConfig cannot remove from existing dedup table");
+ } else {
+ violations.add("dedupConfig cannot add to existing non-dedup table");
+ }
+ } else if (existingDedupEnabled) {
+ DedupConfig existingDedupConfig = existingConfig.getDedupConfig();
+ DedupConfig newDedupConfig = newConfig.getDedupConfig();
+
+ if (existingDedupConfig.getHashFunction() != newDedupConfig.getHashFunction()) {
+ violations.add(String.format("hashFunction (%s -> %s)", existingDedupConfig.getHashFunction(),
+ newDedupConfig.getHashFunction()));
+ }
+
+ if (!Objects.equals(existingDedupConfig.getDedupTimeColumn(), newDedupConfig.getDedupTimeColumn())) {
+ violations.add(String.format("dedupTimeColumn (%s -> %s)", existingDedupConfig.getDedupTimeColumn(),
+ newDedupConfig.getDedupTimeColumn()));
+ }
+ String existingDedupTimeColumn = existingDedupConfig.getDedupTimeColumn();
+ if (existingDedupTimeColumn == null || existingDedupTimeColumn.isEmpty()) {
+ String existingTimeColumn =
+ existingConfig.getValidationConfig() != null ? existingConfig.getValidationConfig().getTimeColumnName()
+ : null;
+ String newTimeColumn =
+ newConfig.getValidationConfig() != null ? newConfig.getValidationConfig().getTimeColumnName() : null;
+ if (!Objects.equals(existingTimeColumn, newTimeColumn)) {
+ violations.add(
+ String.format("timeColumnName (%s -> %s) - used as default dedup time column", existingTimeColumn,
+ newTimeColumn));
+ }
+ }
+ }
+ handleViolations(tableName, violations, force, "dedup");
+ }
+
+ private static void handleViolations(String tableName, List<String> violations, boolean force, String configType) {
+ if (!violations.isEmpty()) {
+ if (force) {
+ LOGGER.warn("Forcing a config: {} update for table {} with changes: {}."
+ + "This may cause data inconsistencies or data loss. Be cautious during compactions, and "
+ + "pause consumption beforehand and disable SNAPSHOT mode in upsertConfig and restart for the changes"
+ + " to kick in. If in doubt, recreate the table with the new configuration.", configType, tableName,
+ violations);
+ } else {
+ throw new IllegalArgumentException(String.format(
Review Comment:
Have to change multiple classes, but added to make this more consistent
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add checks to not update upsert configs after table creation [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #17645:
URL: https://github.com/apache/pinot/pull/17645#issuecomment-3856440417
### :x: 12 Tests Failed:
| Tests completed | Failed | Passed | Skipped |
|---|---|---|---|
| 7504 | 12 | 7492 | 0 |
<details><summary>View the top 3 failed test(s) by shortest run time</summary>
> <pre><code class="language-python">org.apache.pinot.controller.helix.TableCacheTest::testTableCache</code></pre>
> <details><summary>Stack Traces | 0.007s run time</summary>
>
> > <pre><code class="language-python">expected [null] but found [{
> > "schemaName" : "cacheTestTable",
> > "enableColumnBasedNullHandling" : false,
> > "dimensionFieldSpecs" : [ {
> > "name" : "testColumn",
> > "dataType" : "INT",
> > "fieldType" : "DIMENSION"
> > }, {
> > "name" : "$docId",
> > "dataType" : "INT",
> > "fieldType" : "DIMENSION"
> > }, {
> > "name" : "$hostName",
> > "dataType" : "STRING",
> > "fieldType" : "DIMENSION"
> > }, {
> > "name" : "$segmentName",
> > "dataType" : "STRING",
> > "fieldType" : "DIMENSION"
> > }, {
> > "name" : "$partitionId",
> > "dataType" : "STRING",
> > "fieldType" : "DIMENSION",
> > "singleValueField" : false
> > } ]
> > }]</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.controller.api.PinotSchemaRestletResourceTest::testCreateUpdateSchema</code></pre>
> <details><summary>Stack Traces | 0.011s run time</summary>
>
> > <pre><code class="language-python">expected [200] but found [500]</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.controller.api.PinotSchemaRestletResourceTest::testCreateUpdateSchema</code></pre>
> <details><summary>Stack Traces | 0.011s run time</summary>
>
> > <pre><code class="language-python">expected [200] but found [500]</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.controller.api.TableConfigsRestletResourceTest::testListConfigs</code></pre>
> <details><summary>Stack Traces | 0.042s run time</summary>
>
> > <pre><code class="language-python">expected [1] but found [2]</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.controller.api.TableConfigsRestletResourceTest::testListConfigs</code></pre>
> <details><summary>Stack Traces | 0.042s run time</summary>
>
> > <pre><code class="language-python">expected [1] but found [2]</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.controller.api.TableConfigsRestletResourceTest::testUpdateConfig</code></pre>
> <details><summary>Stack Traces | 0.042s run time</summary>
>
> > <pre><code class="language-python">Should fail for trying to PUT config before creating via POST</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.controller.api.TableConfigsRestletResourceTest::testUpdateConfig</code></pre>
> <details><summary>Stack Traces | 0.042s run time</summary>
>
> > <pre><code class="language-python">Should fail for trying to PUT config before creating via POST</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.controller.api.TableConfigsRestletResourceTest::testForceUpdateTableSchemaAndConfigs</code></pre>
> <details><summary>Stack Traces | 0.08s run time</summary>
>
> > <pre><code class="language-python">expected [true] but found [false]</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.controller.api.TableConfigsRestletResourceTest::testForceUpdateTableSchemaAndConfigs</code></pre>
> <details><summary>Stack Traces | 0.08s run time</summary>
>
> > <pre><code class="language-python">expected [true] but found [false]</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.controller.helix.LogicalTableMetadataCacheTest::testLogicalTableCacheWithUpdates</code></pre>
> <details><summary>Stack Traces | 0.122s run time</summary>
>
> > <pre><code class="language-python">org.apache.pinot.common.exception.HttpErrorStatusException: Got error status code: 500 (Internal Server Error) with reason: &quot;Failed to update schema testLogicalTable1. Reason: null&quot; while sending request: /schemas/testLogicalTable1 to controller: runnervmkj6or.net0ibprzlqujb4ppu2xd1znfd.ex.internal.cloudapp.net, version: Unknown</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.controller.helix.LogicalTableMetadataCacheTest::testLogicalTableCacheWithUpdates</code></pre>
> <details><summary>Stack Traces | 0.226s run time</summary>
>
> > <pre><code class="language-python">org.apache.pinot.common.exception.HttpErrorStatusException: Got error status code: 500 (Internal Server Error) with reason: "Failed to update schema testLogicalTable1. Reason: null" while sending request: /schemas/testLogicalTable1 to controller: runnervmkj6or.net0ibprzlqujb4ppu2xd1znfd.ex.internal.cloudapp.net, version: Unknown</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.controller.helix.TableCacheTest::testTableCache</code></pre>
> <details><summary>Stack Traces | 0.228s run time</summary>
>
> > <pre><code class="language-python">No failure message available</code></pre>
>
> </details>
</details>
To view more test analytics, go to the [Test Analytics Dashboard](https://app.codecov.io/gh/apache/pinot/tests/deepthi912%3AdropOutOfOrder-disable)
<sub>📋 Got 3 mins? [Take this short survey](https://forms.gle/22i53Qa1CySZjA6c7) to help us improve Test Analytics.</sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add checks to not update upsert / dedup configs after table creation [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17645:
URL: https://github.com/apache/pinot/pull/17645#discussion_r2875540997
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigsRestletResource.java:
##########
@@ -367,7 +368,10 @@ public ConfigSuccessResponse updateConfig(
@ApiParam(value = "Reload the table if the new schema is backward compatible") @DefaultValue("false")
@QueryParam("reload") boolean reload,
@ApiParam(value = "Force update the table schema") @DefaultValue("false") @QueryParam("forceTableSchemaUpdate")
- boolean forceTableSchemaUpdate, String tableConfigsStr, @Context HttpHeaders headers)
+ boolean forceTableSchemaUpdate,
+ @ApiParam(value = "Force update config changes")
+ @QueryParam("force") @DefaultValue("false") boolean force,
Review Comment:
Do we need this extra param? Seems we should just use `forceTableSchemaUpdate`.
Ideally we should check compatibility before posting the updates
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/Schema.java:
##########
@@ -840,10 +840,20 @@ public void updateBooleanFieldsIfNeeded(Schema oldSchema) {
* Backward compatibility requires
* (1) all columns in oldSchema should be retained.
* (2) all column fieldSpecs should be backward compatible with the old ones.
+ * (3) primary key columns should not be changed (used in dimension tables, upsert, and dedup).
*
* @param oldSchema old schema
*/
public boolean isBackwardCompatibleWith(Schema oldSchema) {
+ // Check primary key columns are not changed
+ List<String> oldPrimaryKeys = oldSchema.getPrimaryKeyColumns();
+ List<String> newPrimaryKeys = getPrimaryKeyColumns();
+ if (oldPrimaryKeys != null && !oldPrimaryKeys.isEmpty()) {
Review Comment:
This basically means we allow setting primary keys (old key doesn't exist). Is this intentionally? If so, please make it clear in the javadoc
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/Schema.java:
##########
@@ -840,10 +840,20 @@ public void updateBooleanFieldsIfNeeded(Schema oldSchema) {
* Backward compatibility requires
* (1) all columns in oldSchema should be retained.
* (2) all column fieldSpecs should be backward compatible with the old ones.
+ * (3) primary key columns should not be changed (used in dimension tables, upsert, and dedup).
*
* @param oldSchema old schema
*/
public boolean isBackwardCompatibleWith(Schema oldSchema) {
+ // Check primary key columns are not changed
+ List<String> oldPrimaryKeys = oldSchema.getPrimaryKeyColumns();
+ List<String> newPrimaryKeys = getPrimaryKeyColumns();
+ if (oldPrimaryKeys != null && !oldPrimaryKeys.isEmpty()) {
Review Comment:
(nit)
```suggestion
if (CollectionUtils.isNotEmpty(oldPrimaryKeys)) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add checks to not update upsert / dedup configs after table creation [pinot]
Posted by "deepthi912 (via GitHub)" <gi...@apache.org>.
deepthi912 commented on code in PR #17645:
URL: https://github.com/apache/pinot/pull/17645#discussion_r2875645892
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/Schema.java:
##########
@@ -840,10 +840,20 @@ public void updateBooleanFieldsIfNeeded(Schema oldSchema) {
* Backward compatibility requires
* (1) all columns in oldSchema should be retained.
* (2) all column fieldSpecs should be backward compatible with the old ones.
+ * (3) primary key columns should not be changed (used in dimension tables, upsert, and dedup).
*
* @param oldSchema old schema
*/
public boolean isBackwardCompatibleWith(Schema oldSchema) {
+ // Check primary key columns are not changed
+ List<String> oldPrimaryKeys = oldSchema.getPrimaryKeyColumns();
+ List<String> newPrimaryKeys = getPrimaryKeyColumns();
+ if (oldPrimaryKeys != null && !oldPrimaryKeys.isEmpty()) {
Review Comment:
I think comparing list is not the right way, so changed it to set actually.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add checks to not update upsert / dedup configs after table creation [pinot]
Posted by "deepthi912 (via GitHub)" <gi...@apache.org>.
deepthi912 commented on code in PR #17645:
URL: https://github.com/apache/pinot/pull/17645#discussion_r2882045803
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1582,6 +1585,14 @@ private void updateSchema(Schema schema, Schema oldSchema, boolean forceTableSch
.append(" is not backward-compatible with the existing schema.");
errorMsg.append("\n\nIncompatibility Details:");
+ // Check for primary key column changes
+ List<String> oldPrimaryKeys = oldSchema.getPrimaryKeyColumns();
+ List<String> newPrimaryKeys = schema.getPrimaryKeyColumns();
+ if (oldPrimaryKeys != null && !oldPrimaryKeys.isEmpty() && !oldPrimaryKeys.equals(newPrimaryKeys)) {
Review Comment:
Made it same in both places. Allowing the user to add in pks might help in making non-upsert table add in upsert configs and pks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add checks to not update upsert / dedup configs after table creation [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17645:
URL: https://github.com/apache/pinot/pull/17645#discussion_r2784602365
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSchemaRestletResource.java:
##########
@@ -483,6 +483,9 @@ private SuccessResponse updateSchema(String schemaName, Schema schema, boolean r
_controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_SCHEMA_UPLOAD_ERROR, 1L);
throw new ControllerApplicationException(LOGGER, String.format("Failed to find table %s to reload", schemaName),
Response.Status.NOT_FOUND, e);
+ } catch (IllegalArgumentException e) {
Review Comment:
This should be modeled as `SchemaBackwardIncompatibleException`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1631,6 +1635,42 @@ private void updateSchema(Schema schema, Schema oldSchema, boolean forceTableSch
LOGGER.info("Updated schema: {}", schemaName);
}
+ /**
+ * Validates that primary key columns are not changed for upsert tables.
+ * Changing primary key columns can lead to data inconsistencies between replicas.
+ *
+ * @param schemaName the name of the schema
+ * @param oldSchema the existing schema
+ * @param newSchema the new schema being applied
+ * @throws IllegalArgumentException if primary key columns are changed and forceUpdate is false
+ */
+ private void validatePrimaryKeyColumnsUpdate(String schemaName, Schema oldSchema, Schema newSchema) {
Review Comment:
We can simplify the logic to just not allowing primary key columns change. Primary key columns are used in dimension table, upsert and dedup. All of them requires special handling
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1570,6 +1572,8 @@ private void updateSchema(Schema schema, Schema oldSchema, boolean forceTableSch
LOGGER.info("New schema: {} is the same as the existing schema, not updating it", schemaName);
return;
}
+
+ validatePrimaryKeyColumnsUpdate(schemaName, oldSchema, schema);
Review Comment:
Integrate this into `Schema.isBackwardCompatibleWith()`
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -1173,6 +1174,156 @@ static void validatePartialUpsertStrategies(TableConfig tableConfig, Schema sche
}
}
+ /**
+ * Validates that critical upsert configuration fields are not changed during table config update.
+ * Checks: mode, hashFunction, comparisonColumns, timeColumn (when no comparison columns),
+ * deleteRecordColumn, dropOutOfOrderRecord, outOfOrderRecordColumn,
+ * partialUpsertStrategies, defaultPartialUpsertStrategy.
+ *
+ * @param existingConfig the existing table config
+ * @param newConfig the new table config being applied
+ * @param force if true, logs a warning instead of throwing an exception
+ * @throws IllegalArgumentException if any critical config field is changed and force is false
+ */
+ public static void validateUpsertConfigUpdate(TableConfig existingConfig, TableConfig newConfig, boolean force) {
Review Comment:
Add a method `void validateBackwardCompatibility(TableConfig newConfig, TableConfig existingConfig)`.
`force` flag should be handled at the caller side.
(minor) To be consistent with schema, put the new config in front of the existing config
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -1173,6 +1174,156 @@ static void validatePartialUpsertStrategies(TableConfig tableConfig, Schema sche
}
}
+ /**
+ * Validates that critical upsert configuration fields are not changed during table config update.
+ * Checks: mode, hashFunction, comparisonColumns, timeColumn (when no comparison columns),
+ * deleteRecordColumn, dropOutOfOrderRecord, outOfOrderRecordColumn,
+ * partialUpsertStrategies, defaultPartialUpsertStrategy.
+ *
+ * @param existingConfig the existing table config
+ * @param newConfig the new table config being applied
+ * @param force if true, logs a warning instead of throwing an exception
+ * @throws IllegalArgumentException if any critical config field is changed and force is false
+ */
+ public static void validateUpsertConfigUpdate(TableConfig existingConfig, TableConfig newConfig, boolean force) {
+ String tableName = existingConfig.getTableName();
+ List<String> violations = new ArrayList<>();
+
+ boolean existingUpsertEnabled = existingConfig.isUpsertEnabled();
+ boolean newUpsertEnabled = newConfig.isUpsertEnabled();
+
+ // Check if upsert is being added or removed
+ if (existingUpsertEnabled != newUpsertEnabled) {
+ if (existingUpsertEnabled) {
+ violations.add("upsertConfig cannot remove from existing upsert table");
+ } else {
+ violations.add("upsertConfig cannot add to existing non-upsert table");
+ }
+ } else if (existingUpsertEnabled) {
+ UpsertConfig existingUpsertConfig = existingConfig.getUpsertConfig();
+ UpsertConfig newUpsertConfig = newConfig.getUpsertConfig();
+
+ if (existingUpsertConfig.getMode() != newUpsertConfig.getMode()) {
+ violations.add(
+ String.format("mode (%s -> %s)", existingUpsertConfig.getMode(), newUpsertConfig.getMode()));
+ }
+ if (existingUpsertConfig.getHashFunction() != newUpsertConfig.getHashFunction()) {
+ violations.add(String.format("hashFunction (%s -> %s)", existingUpsertConfig.getHashFunction(),
+ newUpsertConfig.getHashFunction()));
+ }
+ if (!Objects.equals(existingUpsertConfig.getComparisonColumns(),
+ newUpsertConfig.getComparisonColumns())) {
+ violations.add(
+ String.format("comparisonColumns (%s -> %s)", existingUpsertConfig.getComparisonColumns(),
+ newUpsertConfig.getComparisonColumns()));
+ }
+ List<String> existingComparisonColumns = existingUpsertConfig.getComparisonColumns();
+ if (existingComparisonColumns == null || existingComparisonColumns.isEmpty()) {
+ String existingTimeColumn =
+ existingConfig.getValidationConfig() != null ? existingConfig.getValidationConfig().getTimeColumnName()
+ : null;
+ String newTimeColumn =
+ newConfig.getValidationConfig() != null ? newConfig.getValidationConfig().getTimeColumnName() : null;
+ if (!Objects.equals(existingTimeColumn, newTimeColumn)) {
+ violations.add(
+ String.format("timeColumnName (%s -> %s) - used as default comparison column", existingTimeColumn,
+ newTimeColumn));
+ }
+ }
+ if (existingUpsertConfig.isDropOutOfOrderRecord() != newUpsertConfig.isDropOutOfOrderRecord()) {
+ violations.add(
+ String.format("dropOutOfOrderRecord (%s -> %s)", existingUpsertConfig.isDropOutOfOrderRecord(),
+ newUpsertConfig.isDropOutOfOrderRecord()));
+ }
+ if (!Objects.equals(existingUpsertConfig.getOutOfOrderRecordColumn(),
+ newUpsertConfig.getOutOfOrderRecordColumn())) {
+ violations.add(String.format("outOfOrderRecordColumn (%s -> %s)",
+ existingUpsertConfig.getOutOfOrderRecordColumn(), newUpsertConfig.getOutOfOrderRecordColumn()));
+ }
+ if (existingUpsertConfig.getMode() == UpsertConfig.Mode.PARTIAL) {
+ if (!Objects.equals(existingUpsertConfig.getPartialUpsertStrategies(),
+ newUpsertConfig.getPartialUpsertStrategies())) {
+ violations.add(String.format("partialUpsertStrategies (%s -> %s)",
+ existingUpsertConfig.getPartialUpsertStrategies(), newUpsertConfig.getPartialUpsertStrategies()));
+ }
+ if (existingUpsertConfig.getDefaultPartialUpsertStrategy()
+ != newUpsertConfig.getDefaultPartialUpsertStrategy()) {
+ violations.add(String.format("defaultPartialUpsertStrategy (%s -> %s)",
+ existingUpsertConfig.getDefaultPartialUpsertStrategy(),
+ newUpsertConfig.getDefaultPartialUpsertStrategy()));
+ }
+ }
+ }
+ handleViolations(tableName, violations, force, "upsert");
+ }
+
+ /**
+ * Validates that critical dedup configuration fields are not changed during table config update.
+ * Checks: dedupEnabled, hashFunction, dedupTimeColumn, timeColumnName (when dedupTimeColumn not specified).
+ *
+ * @param existingConfig the existing table config
+ * @param newConfig the new table config being applied
+ * @param force if true, logs a warning instead of throwing an exception
+ * @throws IllegalArgumentException if any critical config field is changed and force is false
+ */
+ public static void validateDedupConfigUpdate(TableConfig existingConfig, TableConfig newConfig, boolean force) {
+ String tableName = existingConfig.getTableName();
+ List<String> violations = new ArrayList<>();
+
+ boolean existingDedupEnabled = existingConfig.isDedupEnabled();
+ boolean newDedupEnabled = newConfig.isDedupEnabled();
+ if (existingDedupEnabled != newDedupEnabled) {
+ if (existingDedupEnabled) {
+ violations.add("dedupConfig cannot remove from existing dedup table");
+ } else {
+ violations.add("dedupConfig cannot add to existing non-dedup table");
+ }
+ } else if (existingDedupEnabled) {
+ DedupConfig existingDedupConfig = existingConfig.getDedupConfig();
+ DedupConfig newDedupConfig = newConfig.getDedupConfig();
+
+ if (existingDedupConfig.getHashFunction() != newDedupConfig.getHashFunction()) {
+ violations.add(String.format("hashFunction (%s -> %s)", existingDedupConfig.getHashFunction(),
+ newDedupConfig.getHashFunction()));
+ }
+
+ if (!Objects.equals(existingDedupConfig.getDedupTimeColumn(), newDedupConfig.getDedupTimeColumn())) {
+ violations.add(String.format("dedupTimeColumn (%s -> %s)", existingDedupConfig.getDedupTimeColumn(),
+ newDedupConfig.getDedupTimeColumn()));
+ }
+ String existingDedupTimeColumn = existingDedupConfig.getDedupTimeColumn();
+ if (existingDedupTimeColumn == null || existingDedupTimeColumn.isEmpty()) {
+ String existingTimeColumn =
+ existingConfig.getValidationConfig() != null ? existingConfig.getValidationConfig().getTimeColumnName()
+ : null;
+ String newTimeColumn =
+ newConfig.getValidationConfig() != null ? newConfig.getValidationConfig().getTimeColumnName() : null;
+ if (!Objects.equals(existingTimeColumn, newTimeColumn)) {
+ violations.add(
+ String.format("timeColumnName (%s -> %s) - used as default dedup time column", existingTimeColumn,
+ newTimeColumn));
+ }
+ }
+ }
+ handleViolations(tableName, violations, force, "dedup");
+ }
+
+ private static void handleViolations(String tableName, List<String> violations, boolean force, String configType) {
+ if (!violations.isEmpty()) {
+ if (force) {
+ LOGGER.warn("Forcing a config: {} update for table {} with changes: {}."
+ + "This may cause data inconsistencies or data loss. Be cautious during compactions, and "
+ + "pause consumption beforehand and disable SNAPSHOT mode in upsertConfig and restart for the changes"
+ + " to kick in. If in doubt, recreate the table with the new configuration.", configType, tableName,
+ violations);
+ } else {
+ throw new IllegalArgumentException(String.format(
Review Comment:
Let's introduce an exception `TableConfigBackwardIncompatibleException`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -737,7 +737,10 @@ private void validateLogicalTableReference(String tableName, TableType tableType
public ConfigSuccessResponse updateTableConfig(
@ApiParam(value = "Name of the table to update", required = true) @PathParam("tableName") String tableName,
@ApiParam(value = "comma separated list of validation type(s) to skip. supported types: (ALL|TASK|UPSERT)")
- @QueryParam("validationTypesToSkip") @Nullable String typesToSkip, @Context HttpHeaders headers,
+ @QueryParam("validationTypesToSkip") @Nullable String typesToSkip,
+ @ApiParam(value = "Force config changes")
+ @QueryParam("forceConfigUpdate") @DefaultValue("false") boolean forceConfigUpdate,
Review Comment:
Make it consistent with schema update
```suggestion
@QueryParam("force") @DefaultValue("false") boolean force,
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add checks to not update upsert / dedup configs after table creation [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang merged PR #17645:
URL: https://github.com/apache/pinot/pull/17645
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add checks to not update upsert / dedup configs after table creation [pinot]
Posted by "deepthi912 (via GitHub)" <gi...@apache.org>.
deepthi912 commented on code in PR #17645:
URL: https://github.com/apache/pinot/pull/17645#discussion_r2875645892
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/Schema.java:
##########
@@ -840,10 +840,20 @@ public void updateBooleanFieldsIfNeeded(Schema oldSchema) {
* Backward compatibility requires
* (1) all columns in oldSchema should be retained.
* (2) all column fieldSpecs should be backward compatible with the old ones.
+ * (3) primary key columns should not be changed (used in dimension tables, upsert, and dedup).
*
* @param oldSchema old schema
*/
public boolean isBackwardCompatibleWith(Schema oldSchema) {
+ // Check primary key columns are not changed
+ List<String> oldPrimaryKeys = oldSchema.getPrimaryKeyColumns();
+ List<String> newPrimaryKeys = getPrimaryKeyColumns();
+ if (oldPrimaryKeys != null && !oldPrimaryKeys.isEmpty()) {
Review Comment:
I think comparing list is might not be the right way, so changed it to set actually.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add checks to not update upsert / dedup configs after table creation [pinot]
Posted by "deepthi912 (via GitHub)" <gi...@apache.org>.
deepthi912 commented on code in PR #17645:
URL: https://github.com/apache/pinot/pull/17645#discussion_r2882023858
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1582,6 +1585,14 @@ private void updateSchema(Schema schema, Schema oldSchema, boolean forceTableSch
.append(" is not backward-compatible with the existing schema.");
errorMsg.append("\n\nIncompatibility Details:");
+ // Check for primary key column changes
+ List<String> oldPrimaryKeys = oldSchema.getPrimaryKeyColumns();
+ List<String> newPrimaryKeys = schema.getPrimaryKeyColumns();
+ if (oldPrimaryKeys != null && !oldPrimaryKeys.isEmpty() && !oldPrimaryKeys.equals(newPrimaryKeys)) {
Review Comment:
This may basically mean changing non-upsert table to upsert table when there are no existing pks. Yeah, maybe let me do that..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add checks to not update upsert / dedup configs after table creation [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17645:
URL: https://github.com/apache/pinot/pull/17645#discussion_r2881051769
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigsRestletResource.java:
##########
@@ -367,7 +368,8 @@ public ConfigSuccessResponse updateConfig(
@ApiParam(value = "Reload the table if the new schema is backward compatible") @DefaultValue("false")
@QueryParam("reload") boolean reload,
@ApiParam(value = "Force update the table schema") @DefaultValue("false") @QueryParam("forceTableSchemaUpdate")
- boolean forceTableSchemaUpdate, String tableConfigsStr, @Context HttpHeaders headers)
+ boolean forceTableSchemaUpdate, @ApiParam(value = "Force update config changes") String tableConfigsStr,
Review Comment:
`@ApiParam` added to the wrong field
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1582,6 +1585,14 @@ private void updateSchema(Schema schema, Schema oldSchema, boolean forceTableSch
.append(" is not backward-compatible with the existing schema.");
errorMsg.append("\n\nIncompatibility Details:");
+ // Check for primary key column changes
+ List<String> oldPrimaryKeys = oldSchema.getPrimaryKeyColumns();
+ List<String> newPrimaryKeys = schema.getPrimaryKeyColumns();
+ if (oldPrimaryKeys != null && !oldPrimaryKeys.isEmpty() && !oldPrimaryKeys.equals(newPrimaryKeys)) {
Review Comment:
This doesn't align with the check in `Schema`. Do you want to allow adding primary keys when there is no existing one?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add checks to not update upsert / dedup configs after table creation [pinot]
Posted by "deepthi912 (via GitHub)" <gi...@apache.org>.
deepthi912 commented on code in PR #17645:
URL: https://github.com/apache/pinot/pull/17645#discussion_r2882023858
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1582,6 +1585,14 @@ private void updateSchema(Schema schema, Schema oldSchema, boolean forceTableSch
.append(" is not backward-compatible with the existing schema.");
errorMsg.append("\n\nIncompatibility Details:");
+ // Check for primary key column changes
+ List<String> oldPrimaryKeys = oldSchema.getPrimaryKeyColumns();
+ List<String> newPrimaryKeys = schema.getPrimaryKeyColumns();
+ if (oldPrimaryKeys != null && !oldPrimaryKeys.isEmpty() && !oldPrimaryKeys.equals(newPrimaryKeys)) {
Review Comment:
This may basically mean changing non-upsert table to upsert table when there are no existing pks. Yeah, maybe let me do that..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add checks to not update upsert / dedup configs after table creation [pinot]
Posted by "deepthi912 (via GitHub)" <gi...@apache.org>.
deepthi912 commented on code in PR #17645:
URL: https://github.com/apache/pinot/pull/17645#discussion_r2876015547
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/Schema.java:
##########
@@ -840,10 +840,20 @@ public void updateBooleanFieldsIfNeeded(Schema oldSchema) {
* Backward compatibility requires
* (1) all columns in oldSchema should be retained.
* (2) all column fieldSpecs should be backward compatible with the old ones.
+ * (3) primary key columns should not be changed (used in dimension tables, upsert, and dedup).
*
* @param oldSchema old schema
*/
public boolean isBackwardCompatibleWith(Schema oldSchema) {
+ // Check primary key columns are not changed
+ List<String> oldPrimaryKeys = oldSchema.getPrimaryKeyColumns();
+ List<String> newPrimaryKeys = getPrimaryKeyColumns();
+ if (oldPrimaryKeys != null && !oldPrimaryKeys.isEmpty()) {
Review Comment:
Actually, comparing it with list makes more sense as order matters, so changed it back to comparing it as a list
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org