tarun11Mavani commented on code in PR #19270: URL: https://github.com/apache/pinot/pull/19270#discussion_r4450968901
########## pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/ConfigMigrationManager.java: ########## @@ -0,0 +1,320 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.controller.helix.core.periodictask; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import javax.annotation.Nullable; +import org.apache.commons.lang3.tuple.ImmutablePair; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.helix.store.zk.ZkHelixPropertyStore; +import org.apache.helix.zookeeper.datamodel.ZNRecord; +import org.apache.pinot.common.metadata.ZKMetadataProvider; +import org.apache.pinot.common.metrics.ControllerMeter; +import org.apache.pinot.common.metrics.ControllerMetrics; +import org.apache.pinot.controller.ControllerConf; +import org.apache.pinot.controller.ControllerConf.ControllerPeriodicTasksConf; +import org.apache.pinot.controller.LeadControllerManager; +import org.apache.pinot.controller.helix.core.PinotHelixResourceManager; +import org.apache.pinot.segment.local.utils.SchemaUtils; +import org.apache.pinot.segment.local.utils.TableConfigUtils; +import org.apache.pinot.spi.config.ConfigUtils; +import org.apache.pinot.spi.config.migration.ConfigMigrationRegistry; +import org.apache.pinot.spi.config.migration.ConfigMigrationUtils; +import org.apache.pinot.spi.config.migration.MigrationResult; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableConfigValidatorRegistry; +import org.apache.pinot.spi.config.table.ingestion.IngestionConfig; +import org.apache.pinot.spi.config.table.ingestion.TransformConfig; +import org.apache.pinot.spi.data.FieldSpec; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.utils.JsonUtils; +import org.apache.pinot.spi.utils.TableConfigDecoratorRegistry; +import org.apache.pinot.spi.utils.builder.TableNameBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/// Inspects stored table definitions and recommends available schema and table-config migrations. +/// By default only warnings and recommendation meters are emitted. Explicitly enabling +/// the cluster config `controller.config.migration.auto.update.enabled` also persists validated candidates with the +/// captured ZooKeeper version and refreshes caches through the resource manager. +/// +/// Schema transforms move to all affected table configs before being removed, using one version-checked transaction. +/// Raw templates are retained for storage; a separate resolved table config is used for validation and +/// runtime side effects. Table leadership and sequential per-run processing are provided by [ControllerPeriodicTask]. +public class ConfigMigrationManager extends ControllerPeriodicTask<ConfigMigrationManager.Context> { + private static final Logger LOGGER = LoggerFactory.getLogger(ConfigMigrationManager.class); + public static final String TASK_NAME = ConfigMigrationManager.class.getSimpleName(); + + private final ConfigMigrationRegistry _migrationRegistry; + private volatile boolean _autoUpdateEnabled = + ControllerPeriodicTasksConf.DEFAULT_CONFIG_MIGRATION_AUTO_UPDATE_ENABLED; + + public ConfigMigrationManager(PinotHelixResourceManager helixResourceManager, + LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics, + ConfigMigrationRegistry migrationRegistry) { + super(TASK_NAME, config.getConfigMigrationFrequencyInSeconds(), config.getConfigMigrationInitialDelaySeconds(), + config.getConfigMigrationCronExpression(), helixResourceManager, leadControllerManager, controllerMetrics); + _migrationRegistry = migrationRegistry; + } + + @Override + public void onChange(Set<String> changedConfigs, Map<String, String> clusterConfigs) { + if (changedConfigs.contains(ControllerPeriodicTasksConf.CONFIG_MIGRATION_AUTO_UPDATE_ENABLED)) { + _autoUpdateEnabled = Boolean.parseBoolean(clusterConfigs.getOrDefault( + ControllerPeriodicTasksConf.CONFIG_MIGRATION_AUTO_UPDATE_ENABLED, + Boolean.toString(ControllerPeriodicTasksConf.DEFAULT_CONFIG_MIGRATION_AUTO_UPDATE_ENABLED))); + } + } + + /// Tracks shared schemas so a hybrid table's OFFLINE/REALTIME halves report and update it once per run. + public static class Context { + private final Set<String> _processedSchemas = new HashSet<>(); + private final Set<String> _processedTables = new HashSet<>(); + } + + @Override + protected Context preprocess(Properties periodicTaskProperties) { + return new Context(); + } + + @Override + protected void processTable(String tableNameWithType, Context context) { + if (_migrationRegistry.getCurrentTableConfigVersion() == 0 && _migrationRegistry.getCurrentSchemaVersion() == 0) { + return; + } + ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore(); + String schemaName = TableNameBuilder.extractRawTableName(tableNameWithType); + ImmutablePair<Schema, Integer> schemaAndVersion = + ZKMetadataProvider.getSchemaWithVersion(propertyStore, schemaName); + Schema schema = schemaAndVersion != null ? schemaAndVersion.getLeft() : null; + if (_migrationRegistry.getCurrentSchemaVersion() > 0 && schemaAndVersion != null + && context._processedSchemas.add(schemaName)) { + schema = migrateSchema(schema, schemaAndVersion.getRight(), context); + } + if (_migrationRegistry.getCurrentTableConfigVersion() > 0 && context._processedTables.add(tableNameWithType)) { + migrateTableConfig(propertyStore, tableNameWithType, schema); + } + } + + private void migrateTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String tableNameWithType, + @Nullable Schema schema) { + try { + ImmutablePair<TableConfig, Integer> configAndVersion = + ZKMetadataProvider.getTableConfigWithVersion(propertyStore, tableNameWithType, false, false); + if (configAndVersion == null) { + return; + } + TableConfig original = configAndVersion.getLeft(); + MigrationResult<TableConfig> result = _migrationRegistry.migrateTableConfig(original); + if (!result.isChanged()) { + return; + } + LOGGER.warn("Table: {} has legacy configuration; recommend updating to config migration version: {}. " + + "Automatic updates enabled: {}", tableNameWithType, result.getVersion(), _autoUpdateEnabled); + _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.CONFIG_MIGRATION_RECOMMENDED, 1L); + if (schema == null) { + return; + } + TableConfig migrated = result.getConfig(); + TableConfig effective = TableConfigDecoratorRegistry.applyDecorator( + ConfigUtils.applyConfigWithEnvVariablesAndSystemProperties(migrated)); + TableConfig existingEffective = TableConfigDecoratorRegistry.applyDecorator( + ConfigUtils.applyConfigWithEnvVariablesAndSystemProperties(original)); + TableConfigUtils.validate(effective, schema, null, existingEffective); + TableConfigValidatorRegistry.validate(effective, schema); + if (!_autoUpdateEnabled) { + return; + } + _pinotHelixResourceManager.setExistingTableConfig(migrated, effective, configAndVersion.getRight(), false); + LOGGER.info("Updated table config for table: {} to migration version: {}", tableNameWithType, + result.getVersion()); + _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.CONFIG_MIGRATION_SUCCESS, 1L); + } catch (Exception e) { + // Exception messages can contain resolved config values; report only the type, never the config payload. + LOGGER.warn("Could not complete table config migration for table: {}; will recheck next scan ({})", + tableNameWithType, e.getClass().getSimpleName()); + _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.CONFIG_MIGRATION_FAILURE, 1L); + } + } + + private Schema migrateSchema(Schema original, int expectedVersion, Context context) { + String schemaName = original.getSchemaName(); + try { + if (ConfigMigrationUtils.getSchemaVersion(original) > _migrationRegistry.getCurrentSchemaVersion()) { + return original; + } + MigrationResult<Schema> result = _migrationRegistry.migrateSchema(original); + Schema migrated = result.getConfig(); + Map<String, String> transforms = getSchemaTransforms(migrated); + if (!result.isChanged() && transforms.isEmpty()) { + return original; + } + LOGGER.warn("Schema: {} has legacy configuration; recommend updating to schema migration version: {}. " + + "Automatic updates enabled: {}", schemaName, result.getVersion(), _autoUpdateEnabled); + _controllerMetrics.addMeteredTableValue(schemaName, ControllerMeter.SCHEMA_MIGRATION_RECOMMENDED, 1L); + if (!transforms.isEmpty()) { + return migrateSchemaTransforms(original, migrated, expectedVersion, transforms, context); + } + List<TableConfig> tableConfigs = new ArrayList<>(); + for (String tableName : _pinotHelixResourceManager.getExistingTableNamesWithType(schemaName, null)) { + TableConfig config = _pinotHelixResourceManager.getTableConfig(tableName); + if (config != null) { + tableConfigs.add(config); + } + } + SchemaUtils.validate(migrated, tableConfigs, false, original); + if (!_autoUpdateEnabled) { + return original; + } + if (!_pinotHelixResourceManager.updateSchema(migrated, expectedVersion)) { + _controllerMetrics.addMeteredTableValue(schemaName, ControllerMeter.CONFIG_MIGRATION_FAILURE, 1L); + return original; + } + LOGGER.info("Updated schema: {} to migration version: {}", schemaName, result.getVersion()); + _controllerMetrics.addMeteredTableValue(schemaName, ControllerMeter.CONFIG_MIGRATION_SUCCESS, 1L); + return migrated; + } catch (Exception e) { + LOGGER.warn("Could not complete schema migration: {}; will recheck next scan ({})", + schemaName, e.getClass().getSimpleName()); + _controllerMetrics.addMeteredTableValue(schemaName, ControllerMeter.CONFIG_MIGRATION_FAILURE, 1L); + return original; + } + } + + // Read legacy schema transforms deliberately; empty and virtual transforms have different runtime semantics. + @SuppressWarnings("deprecation") + private static Map<String, String> getSchemaTransforms(Schema schema) { + Map<String, String> transforms = new LinkedHashMap<>(); + for (FieldSpec field : schema.getAllFieldSpecs()) { + String transform = field.getTransformFunction(); + if (!field.isVirtualColumn() && transform != null && !transform.isEmpty()) { + transforms.put(field.getName(), transform); + } + } + return transforms; + } + + // Clear deprecated schema properties only in a copy whose table replacements are committed atomically. + @SuppressWarnings("deprecation") + private Schema migrateSchemaTransforms(Schema original, Schema candidate, int expectedVersion, + Map<String, String> transforms, Context context) + throws Exception { + String schemaName = original.getSchemaName(); + Schema migrated = Schema.fromString(candidate.toSingleLineJsonString()); + for (String column : transforms.keySet()) { + migrated.getFieldSpecFor(column).setTransformFunction(null); + } + ConfigMigrationUtils.setSchemaVersion(migrated, _migrationRegistry.getCurrentSchemaVersion()); + List<String> tables = _pinotHelixResourceManager.getExistingTableNamesWithType(schemaName, null); + context._processedTables.addAll(tables); + Map<String, Pair<TableConfig, Integer>> tableConfigs = new LinkedHashMap<>(); + List<TableConfig> effectiveConfigs = new ArrayList<>(); + Set<String> changedTables = new HashSet<>(); + for (String tableName : tables) { + Pair<TableConfig, Integer> stored = ZKMetadataProvider.getTableConfigWithVersion( + _pinotHelixResourceManager.getPropertyStore(), tableName, false, false); + if (stored == null || ConfigMigrationUtils.getTableConfigVersion(stored.getLeft()) + > _migrationRegistry.getCurrentTableConfigVersion()) { + throw new IllegalStateException("Table config unavailable for schema transform migration"); + } + TableConfig existing = stored.getLeft(); + TableConfig updated = new TableConfig(_migrationRegistry.migrateTableConfig(existing).getConfig()); Review Comment: new TableConfig(registry.migrateTableConfig(existing).getConfig())` is a shallow copy. When the migration chain changes nothing, the registry returns the same instance as `existing`, and `TableConfig(TableConfig)` reuses the same `IngestionConfig` (TableConfig.java:250). `addSchemaTransforms` then calls `ingestion.setTransformConfigs(updated)` on that shared object, so it mutates the stored config in place. Effects: - `changed` (l.246) is false, so the table gets no `CONFIG_MIGRATION_RECOMMENDED` and no version marker. - `existingEffective` / `expectedEffective` (l.252-258) are built from the already-mutated config, so the guard at l.263 cannot catch a decorator that rewrites the moved entry. - `TableConfigUtils.validate` (l.266) receives a baseline that already contains the moved transform. Repro: a modern table (no legacy fields) with a non-null `ingestionConfig` holding some other transform, plus a schema `transformFunction` on a column that is not yet in `transformConfigs`. With auto-update off, only `SCHEMA_MIGRATION_RECOMMENDED` is emitted. The same scenario with `ingestionConfig == null` does emit the table recommendation. Suggest deep-copying before mutating (JSON round trip, as the registry already does), and computing `changed` against an untouched snapshot. ########## pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java: ########## @@ -2103,6 +2216,51 @@ public void addTable(TableConfig tableConfig, List<StreamMetadata> streamMetadat LOGGER.info("Adding table {}: Successfully added table", tableNameWithType); } + /// Fences schema cleanup against new physical tables by changing the schema data version in the same transaction. + /// The exact stored record is retained, including fields unknown to this controller. + @VisibleForTesting + void createTableConfigWithSchemaVersion(TableConfig tableConfig) + throws IOException { + String tableName = tableConfig.getTableName(); + String schemaPath = ZKMetadataProvider.constructPropertyStorePathForSchema( + TableNameBuilder.extractRawTableName(tableName)); + String tablePath = ZKMetadataProvider.constructPropertyStorePathForResourceConfig(tableName); + String tableParent = tablePath.substring(0, tablePath.lastIndexOf('/')); + for (int attempt = 0; attempt < 3; attempt++) { + Stat stat = new Stat(); + ZNRecord schemaRecord = _propertyStore.get(schemaPath, stat, AccessOption.PERSISTENT); + if (schemaRecord == null) { + throw new InvalidTableConfigException("No schema defined for table: " + tableName); Review Comment: `createTableConfigWithSchemaVersion` only checks that the raw schema ZNRecord exists. The previous `addTable` guard, `ZKMetadataProvider.getSchema(_propertyStore, rawTableName) == null` -> `InvalidTableConfigException`, is removed in this PR, so a stored schema that cannot be parsed is no longer rejected. Suggest also running `SchemaSerDeUtils.fromZNRecord(schemaRecord)` in a try/catch and throwing `InvalidTableConfigException`, while still writing back the raw record so unknown fields are preserved. ########## pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/ConfigMigrationManager.java: ########## @@ -0,0 +1,320 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.controller.helix.core.periodictask; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import javax.annotation.Nullable; +import org.apache.commons.lang3.tuple.ImmutablePair; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.helix.store.zk.ZkHelixPropertyStore; +import org.apache.helix.zookeeper.datamodel.ZNRecord; +import org.apache.pinot.common.metadata.ZKMetadataProvider; +import org.apache.pinot.common.metrics.ControllerMeter; +import org.apache.pinot.common.metrics.ControllerMetrics; +import org.apache.pinot.controller.ControllerConf; +import org.apache.pinot.controller.ControllerConf.ControllerPeriodicTasksConf; +import org.apache.pinot.controller.LeadControllerManager; +import org.apache.pinot.controller.helix.core.PinotHelixResourceManager; +import org.apache.pinot.segment.local.utils.SchemaUtils; +import org.apache.pinot.segment.local.utils.TableConfigUtils; +import org.apache.pinot.spi.config.ConfigUtils; +import org.apache.pinot.spi.config.migration.ConfigMigrationRegistry; +import org.apache.pinot.spi.config.migration.ConfigMigrationUtils; +import org.apache.pinot.spi.config.migration.MigrationResult; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableConfigValidatorRegistry; +import org.apache.pinot.spi.config.table.ingestion.IngestionConfig; +import org.apache.pinot.spi.config.table.ingestion.TransformConfig; +import org.apache.pinot.spi.data.FieldSpec; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.utils.JsonUtils; +import org.apache.pinot.spi.utils.TableConfigDecoratorRegistry; +import org.apache.pinot.spi.utils.builder.TableNameBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/// Inspects stored table definitions and recommends available schema and table-config migrations. +/// By default only warnings and recommendation meters are emitted. Explicitly enabling +/// the cluster config `controller.config.migration.auto.update.enabled` also persists validated candidates with the +/// captured ZooKeeper version and refreshes caches through the resource manager. +/// +/// Schema transforms move to all affected table configs before being removed, using one version-checked transaction. +/// Raw templates are retained for storage; a separate resolved table config is used for validation and +/// runtime side effects. Table leadership and sequential per-run processing are provided by [ControllerPeriodicTask]. +public class ConfigMigrationManager extends ControllerPeriodicTask<ConfigMigrationManager.Context> { + private static final Logger LOGGER = LoggerFactory.getLogger(ConfigMigrationManager.class); + public static final String TASK_NAME = ConfigMigrationManager.class.getSimpleName(); + + private final ConfigMigrationRegistry _migrationRegistry; + private volatile boolean _autoUpdateEnabled = + ControllerPeriodicTasksConf.DEFAULT_CONFIG_MIGRATION_AUTO_UPDATE_ENABLED; + + public ConfigMigrationManager(PinotHelixResourceManager helixResourceManager, + LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics, + ConfigMigrationRegistry migrationRegistry) { + super(TASK_NAME, config.getConfigMigrationFrequencyInSeconds(), config.getConfigMigrationInitialDelaySeconds(), + config.getConfigMigrationCronExpression(), helixResourceManager, leadControllerManager, controllerMetrics); + _migrationRegistry = migrationRegistry; + } + + @Override + public void onChange(Set<String> changedConfigs, Map<String, String> clusterConfigs) { + if (changedConfigs.contains(ControllerPeriodicTasksConf.CONFIG_MIGRATION_AUTO_UPDATE_ENABLED)) { + _autoUpdateEnabled = Boolean.parseBoolean(clusterConfigs.getOrDefault( + ControllerPeriodicTasksConf.CONFIG_MIGRATION_AUTO_UPDATE_ENABLED, + Boolean.toString(ControllerPeriodicTasksConf.DEFAULT_CONFIG_MIGRATION_AUTO_UPDATE_ENABLED))); + } + } + + /// Tracks shared schemas so a hybrid table's OFFLINE/REALTIME halves report and update it once per run. + public static class Context { + private final Set<String> _processedSchemas = new HashSet<>(); + private final Set<String> _processedTables = new HashSet<>(); + } + + @Override + protected Context preprocess(Properties periodicTaskProperties) { + return new Context(); + } + + @Override + protected void processTable(String tableNameWithType, Context context) { + if (_migrationRegistry.getCurrentTableConfigVersion() == 0 && _migrationRegistry.getCurrentSchemaVersion() == 0) { + return; + } + ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore(); + String schemaName = TableNameBuilder.extractRawTableName(tableNameWithType); + ImmutablePair<Schema, Integer> schemaAndVersion = + ZKMetadataProvider.getSchemaWithVersion(propertyStore, schemaName); + Schema schema = schemaAndVersion != null ? schemaAndVersion.getLeft() : null; + if (_migrationRegistry.getCurrentSchemaVersion() > 0 && schemaAndVersion != null + && context._processedSchemas.add(schemaName)) { + schema = migrateSchema(schema, schemaAndVersion.getRight(), context); + } + if (_migrationRegistry.getCurrentTableConfigVersion() > 0 && context._processedTables.add(tableNameWithType)) { + migrateTableConfig(propertyStore, tableNameWithType, schema); + } + } + + private void migrateTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String tableNameWithType, + @Nullable Schema schema) { + try { + ImmutablePair<TableConfig, Integer> configAndVersion = + ZKMetadataProvider.getTableConfigWithVersion(propertyStore, tableNameWithType, false, false); + if (configAndVersion == null) { + return; + } + TableConfig original = configAndVersion.getLeft(); + MigrationResult<TableConfig> result = _migrationRegistry.migrateTableConfig(original); + if (!result.isChanged()) { + return; + } + LOGGER.warn("Table: {} has legacy configuration; recommend updating to config migration version: {}. " + + "Automatic updates enabled: {}", tableNameWithType, result.getVersion(), _autoUpdateEnabled); + _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.CONFIG_MIGRATION_RECOMMENDED, 1L); + if (schema == null) { + return; + } + TableConfig migrated = result.getConfig(); + TableConfig effective = TableConfigDecoratorRegistry.applyDecorator( + ConfigUtils.applyConfigWithEnvVariablesAndSystemProperties(migrated)); + TableConfig existingEffective = TableConfigDecoratorRegistry.applyDecorator( + ConfigUtils.applyConfigWithEnvVariablesAndSystemProperties(original)); + TableConfigUtils.validate(effective, schema, null, existingEffective); + TableConfigValidatorRegistry.validate(effective, schema); + if (!_autoUpdateEnabled) { + return; + } + _pinotHelixResourceManager.setExistingTableConfig(migrated, effective, configAndVersion.getRight(), false); + LOGGER.info("Updated table config for table: {} to migration version: {}", tableNameWithType, + result.getVersion()); + _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.CONFIG_MIGRATION_SUCCESS, 1L); + } catch (Exception e) { + // Exception messages can contain resolved config values; report only the type, never the config payload. + LOGGER.warn("Could not complete table config migration for table: {}; will recheck next scan ({})", + tableNameWithType, e.getClass().getSimpleName()); + _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.CONFIG_MIGRATION_FAILURE, 1L); + } + } + + private Schema migrateSchema(Schema original, int expectedVersion, Context context) { + String schemaName = original.getSchemaName(); + try { + if (ConfigMigrationUtils.getSchemaVersion(original) > _migrationRegistry.getCurrentSchemaVersion()) { + return original; + } + MigrationResult<Schema> result = _migrationRegistry.migrateSchema(original); + Schema migrated = result.getConfig(); + Map<String, String> transforms = getSchemaTransforms(migrated); + if (!result.isChanged() && transforms.isEmpty()) { + return original; + } + LOGGER.warn("Schema: {} has legacy configuration; recommend updating to schema migration version: {}. " + + "Automatic updates enabled: {}", schemaName, result.getVersion(), _autoUpdateEnabled); + _controllerMetrics.addMeteredTableValue(schemaName, ControllerMeter.SCHEMA_MIGRATION_RECOMMENDED, 1L); + if (!transforms.isEmpty()) { + return migrateSchemaTransforms(original, migrated, expectedVersion, transforms, context); + } + List<TableConfig> tableConfigs = new ArrayList<>(); + for (String tableName : _pinotHelixResourceManager.getExistingTableNamesWithType(schemaName, null)) { + TableConfig config = _pinotHelixResourceManager.getTableConfig(tableName); + if (config != null) { + tableConfigs.add(config); + } + } + SchemaUtils.validate(migrated, tableConfigs, false, original); + if (!_autoUpdateEnabled) { + return original; + } + if (!_pinotHelixResourceManager.updateSchema(migrated, expectedVersion)) { + _controllerMetrics.addMeteredTableValue(schemaName, ControllerMeter.CONFIG_MIGRATION_FAILURE, 1L); + return original; + } + LOGGER.info("Updated schema: {} to migration version: {}", schemaName, result.getVersion()); + _controllerMetrics.addMeteredTableValue(schemaName, ControllerMeter.CONFIG_MIGRATION_SUCCESS, 1L); + return migrated; + } catch (Exception e) { + LOGGER.warn("Could not complete schema migration: {}; will recheck next scan ({})", + schemaName, e.getClass().getSimpleName()); + _controllerMetrics.addMeteredTableValue(schemaName, ControllerMeter.CONFIG_MIGRATION_FAILURE, 1L); + return original; + } + } + + // Read legacy schema transforms deliberately; empty and virtual transforms have different runtime semantics. + @SuppressWarnings("deprecation") + private static Map<String, String> getSchemaTransforms(Schema schema) { + Map<String, String> transforms = new LinkedHashMap<>(); + for (FieldSpec field : schema.getAllFieldSpecs()) { + String transform = field.getTransformFunction(); + if (!field.isVirtualColumn() && transform != null && !transform.isEmpty()) { + transforms.put(field.getName(), transform); + } + } + return transforms; + } + + // Clear deprecated schema properties only in a copy whose table replacements are committed atomically. + @SuppressWarnings("deprecation") + private Schema migrateSchemaTransforms(Schema original, Schema candidate, int expectedVersion, + Map<String, String> transforms, Context context) + throws Exception { + String schemaName = original.getSchemaName(); + Schema migrated = Schema.fromString(candidate.toSingleLineJsonString()); + for (String column : transforms.keySet()) { + migrated.getFieldSpecFor(column).setTransformFunction(null); + } + ConfigMigrationUtils.setSchemaVersion(migrated, _migrationRegistry.getCurrentSchemaVersion()); + List<String> tables = _pinotHelixResourceManager.getExistingTableNamesWithType(schemaName, null); + context._processedTables.addAll(tables); Review Comment: `context._processedTables.addAll(tables)` runs before any step that can fail. If the schema-transform plan then throws (for example one table config is unreadable, or validation rejects the move), `processTable` finds those tables already in `_processedTables` (l.117) and skips `migrateTableConfig` for all of them, on every scan. Repro: legacy OFFLINE table (legacy `segmentPushType`), a schema `transformFunction`, and a REALTIME half listed by `getExistingTableNamesWithType` whose config cannot be read. The plan throws "Table config unavailable for schema transform migration" and `CONFIG_MIGRATION_FAILURE` is emitted. `setExistingTableConfig` is never called for the legacy OFFLINE table, even with auto-update on. This contradicts the doc's "other rules retain their independent update paths". Suggest adding to `_processedTables` only after the plan succeeds, or falling through to `migrateTableConfig` when the plan fails. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
