You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/10/16 20:52:35 UTC
(pinot) 11/14: Send the table config and schema refresh message when a schema is overridden (#19677)
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch release-1.6.0-rc
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 223c39af83939cfdcb8b039f513457e9add34e9f
Author: Xiaotian (Jackie) Jiang <17...@users.noreply.github.com>
AuthorDate: Fri Sep 25 17:52:04 2026 -0700
Send the table config and schema refresh message when a schema is overridden (#19677)
(cherry picked from commit 03d7ffd8f4d9def62143009d421526d141de1f98)
(cherry picked from commit 90ee286569d14d1d6322ba9cfb3f7565b42f9996)
---
.../helix/core/PinotHelixResourceManager.java | 50 ++++++++++++++--------
.../PinotHelixResourceManagerStatelessTest.java | 32 ++++++++++++++
2 files changed, 64 insertions(+), 18 deletions(-)
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
index 89a6f72cea6..62d25ed4440 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
@@ -1678,6 +1678,7 @@ public class PinotHelixResourceManager {
// Update existing schema
if (override) {
updateSchema(schema, oldSchema, force);
+ refreshTablesUsingSchema(schemaName);
} else {
throw new SchemaAlreadyExistsException("Schema: " + schemaName + " already exists");
}
@@ -1710,30 +1711,43 @@ public class PinotHelixResourceManager {
}
updateSchema(schema, oldSchema, forceTableSchemaUpdate);
+ if (reload) {
+ reloadTablesUsingSchema(schemaName);
+ } else {
+ refreshTablesUsingSchema(schemaName);
+ }
+ }
+
+ /// Reloads all segments of the tables using the given schema. Logical table schemas are skipped because no
+ /// segments are backed by them.
+ private void reloadTablesUsingSchema(String schemaName)
+ throws TableNotFoundException {
if (ZKMetadataProvider.isLogicalTableExists(_propertyStore, schemaName)) {
- // For logical table schemas, we do not need to reload segments or send schema refresh messages
- LOGGER.info("Logical table schema: {} updated, no need to reload segments or send schema refresh messages",
- schemaName);
+ LOGGER.info("Logical table schema: {} updated, no need to reload segments", schemaName);
return;
}
+ LOGGER.info("Reloading tables with name: {}", schemaName);
+ for (String tableNameWithType : getExistingTableNamesWithType(schemaName, null)) {
+ reloadAllSegments(tableNameWithType, false, null);
+ }
+ }
+
+ /// Sends the table config and schema refresh message to the servers hosting the tables using the given schema.
+ /// Servers reuse their cached schema for ordinary segment loads and only refresh it on this message, an explicit
+ /// reload or a table config update, so every schema update has to send it or newly loaded segments keep being
+ /// processed against the previous schema. Logical table schemas are skipped because no segments are backed by them.
+ private void refreshTablesUsingSchema(String schemaName) {
+ if (ZKMetadataProvider.isLogicalTableExists(_propertyStore, schemaName)) {
+ LOGGER.info("Logical table schema: {} updated, no need to send schema refresh messages", schemaName);
+ return;
+ }
+ LOGGER.info("Refreshing schema for tables with name: {}", schemaName);
try {
- List<String> tableNamesWithType = getExistingTableNamesWithType(schemaName, null);
- if (reload) {
- LOGGER.info("Reloading tables with name: {}", schemaName);
- for (String tableNameWithType : tableNamesWithType) {
- reloadAllSegments(tableNameWithType, false, null);
- }
- } else {
- LOGGER.info("Refreshing schema for tables with name: {}", schemaName);
- for (String tableNameWithType : tableNamesWithType) {
- sendTableConfigSchemaRefreshMessage(tableNameWithType);
- }
+ for (String tableNameWithType : getExistingTableNamesWithType(schemaName, null)) {
+ sendTableConfigSchemaRefreshMessage(tableNameWithType);
}
} catch (TableNotFoundException e) {
- if (reload) {
- throw e;
- }
- // We don't throw exception if no tables found for schema when reload is false. Since this could be valid case
+ // A schema can exist before any table uses it
LOGGER.warn("No tables found for schema (refresh only): {}", schemaName, e);
}
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerStatelessTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerStatelessTest.java
index 64a8ad22274..d2d2dbc757f 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerStatelessTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerStatelessTest.java
@@ -77,8 +77,10 @@ import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.tenant.Tenant;
import org.apache.pinot.spi.config.tenant.TenantRole;
import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DimensionFieldSpec;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.LogicalTableConfig;
+import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.PartitionGroupConsumptionStatus;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
@@ -1131,6 +1133,36 @@ public class PinotHelixResourceManagerStatelessTest extends ControllerTest {
}
}
+ @Test
+ public void testAddSchemaWithOverrideRefreshesServerCaches()
+ throws Exception {
+ String rawTableName = "schemaOverrideRefreshTest";
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ addDummySchema(rawTableName);
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+ .setTableName(rawTableName)
+ .setBrokerTenant(BROKER_TENANT_NAME)
+ .setServerTenant(SERVER_TENANT_NAME)
+ .build();
+ waitForEVToDisappear(tableConfig.getTableName());
+ _helixResourceManager.addTable(tableConfig);
+
+ PinotHelixResourceManager resourceManager = spy(_helixResourceManager);
+ doNothing().when(resourceManager).sendTableConfigSchemaRefreshMessage(offlineTableName);
+
+ try {
+ Schema schema = createDummySchema(rawTableName);
+ schema.addField(new DimensionFieldSpec("dimC", FieldSpec.DataType.STRING, true));
+ resourceManager.addSchema(schema, true, false);
+
+ assertEquals(resourceManager.getSchema(rawTableName), schema);
+ verify(resourceManager).sendTableConfigSchemaRefreshMessage(offlineTableName);
+ } finally {
+ _helixResourceManager.deleteOfflineTable(rawTableName);
+ deleteSchema(rawTableName);
+ }
+ }
+
/// Tests the code path where a subset of merged segments (from the original segmentsTo list)
/// is passed to the endReplace API.
/// @throws Exception
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org