[Feature] SSE Materialized View Creation and Ingestion

pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java

@@ -84,6 +84,9 @@ private ZKMetadataProvider() {
   private static final String PROPERTYSTORE_CLUSTER_CONFIGS_PREFIX = "/CONFIGS/CLUSTER";
   private static final String PROPERTYSTORE_SEGMENT_LINEAGE = "/SEGMENT_LINEAGE";
   private static final String PROPERTYSTORE_MINION_TASK_METADATA_PREFIX = "/MINION_TASK_METADATA";
+  private static final String PROPERTYSTORE_MATERIALIZED_VIEW_DEFINITION_PREFIX =
+      "/CONFIGS/MATERIALIZED_VIEW/DEFINITION";
+  private static final String PROPERTYSTORE_MATERIALIZED_VIEW_RUNTIME_PREFIX = "/CONFIGS/MATERIALIZED_VIEW/RUNTIME";
   private static final String PROPERTYSTORE_QUERY_WORKLOAD_CONFIGS_PREFIX = "/CONFIGS/QUERYWORKLOAD";
   private static final String PROPERTYSTORE_TASK_LOCK_SUFFIX = "-Lock";
 
@@ -328,6 +331,23 @@ public static String constructPropertyStorePathForMinionTaskMetadataDeprecated(S
     return StringUtil.join("/", PROPERTYSTORE_MINION_TASK_METADATA_PREFIX, taskType, tableNameWithType);
   }
 
+  public static String getPropertyStorePathForMaterializedViewDefinitionPrefix() {
+    return PROPERTYSTORE_MATERIALIZED_VIEW_DEFINITION_PREFIX;
+  }
+
+  public static String constructPropertyStorePathForMaterializedViewDefinition(
+      String materializedViewTableNameWithType) {
+    return StringUtil.join("/", PROPERTYSTORE_MATERIALIZED_VIEW_DEFINITION_PREFIX, materializedViewTableNameWithType);
+  }
+
+  public static String getPropertyStorePathForMaterializedViewRuntimePrefix() {
+    return PROPERTYSTORE_MATERIALIZED_VIEW_RUNTIME_PREFIX;
+  }
+
+  public static String constructPropertyStorePathForMaterializedViewRuntime(String materializedViewTableNameWithType) {
+    return StringUtil.join("/", PROPERTYSTORE_MATERIALIZED_VIEW_RUNTIME_PREFIX, materializedViewTableNameWithType);
+  }
+
   public static String constructPropertyStorePathForLogical(String tableName) {
     return StringUtil.join("/", ZkPaths.LOGICAL_TABLE_PARENT_PATH, tableName);
   }

pinot-controller/pom.xml

@@ -36,6 +36,10 @@
     <npmRepositoryUrl>https://registry.npmjs.org/npm/-/</npmRepositoryUrl>
   </properties>
   <dependencies>
+    <dependency>
+      <groupId>org.apache.pinot</groupId>
+      <artifactId>pinot-materialized-view</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.pinot</groupId>
       <artifactId>pinot-query-planner</artifactId>

pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java

@@ -47,6 +47,7 @@
 import org.apache.hc.client5.http.io.HttpClientConnectionManager;
 import org.apache.hc.core5.http.io.SocketConfig;
 import org.apache.hc.core5.util.Timeout;
+import org.apache.helix.HelixAdmin;
 import org.apache.helix.HelixManager;
 import org.apache.helix.HelixManagerFactory;
 import org.apache.helix.InstanceType;
@@ -55,9 +56,11 @@
 import org.apache.helix.manager.zk.ZKHelixManager;
 import org.apache.helix.model.ClusterConstraints;
 import org.apache.helix.model.ConstraintItem;
+import org.apache.helix.model.HelixConfigScope;
 import org.apache.helix.model.InstanceConfig;
 import org.apache.helix.model.MasterSlaveSMD;
 import org.apache.helix.model.Message;
+import org.apache.helix.model.builder.HelixConfigScopeBuilder;
 import org.apache.helix.store.zk.ZkHelixPropertyStore;
 import org.apache.helix.task.TaskDriver;
 import org.apache.helix.zookeeper.constant.ZkSystemPropertyKeys;
@@ -144,6 +147,7 @@
 import org.apache.pinot.core.transport.grpc.GrpcQueryServer;
 import org.apache.pinot.core.util.ListenerConfigUtil;
 import org.apache.pinot.core.util.trace.ContinuousJfrStarter;
+import org.apache.pinot.materializedview.consistency.MaterializedViewConsistencyManager;
 import org.apache.pinot.segment.local.utils.TableConfigUtils;
 import org.apache.pinot.segment.spi.partition.PartitionFunctionFactory;
 import org.apache.pinot.spi.config.instance.InstanceConfigValidatorRegistry;
@@ -239,6 +243,7 @@ public abstract class BaseControllerStarter implements ServiceStartable {
   protected RebalancePreChecker _rebalancePreChecker;
   protected TableRebalanceManager _tableRebalanceManager;
   protected DefaultClusterConfigChangeHandler _clusterConfigChangeHandler;
+  protected MaterializedViewConsistencyManager _materializedViewConsistencyManager;
 
   @Override
   public void init(PinotConfiguration pinotConfiguration)
@@ -605,7 +610,35 @@ private void setUpPinotController() {
     LOGGER.info("Starting Pinot Helix resource manager and connecting to Zookeeper");
     _helixResourceManager.start(_helixParticipantManager, _controllerMetrics);
 
-    // Initialize segment lifecycle event listeners
+    // Register MV consistency manager BEFORE any other lifecycle listener initialization.
+    // PinotHelixResourceManager.notifyMaterializedView* methods are entered as soon as
+    // _helixResourceManager.start() returns (segment add/delete/replace handlers no-op if
+    // the manager is null), so we want this to be the very first thing wired up so any
+    // segment events arriving immediately after Helix participant becomes online will
+    // correctly trigger STALE marking.
+    LOGGER.info("Initializing MaterializedView consistency manager");
+    _materializedViewConsistencyManager = new MaterializedViewConsistencyManager();
+    _materializedViewConsistencyManager.init(_helixResourceManager.getPropertyStore());
+    // Wire a live cluster-config reader so caps like the consistency-manager debounce window
+    // can be overridden via `pinot-admin.sh ClusterConfig` without a controller restart.
+    final HelixAdmin helixAdminForMv = _helixResourceManager.getHelixAdmin();
+    final String helixClusterName = _helixResourceManager.getHelixClusterName();
+    _materializedViewConsistencyManager.setClusterConfigReader(configName -> {
+      try {
+        HelixConfigScope scope = new HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.CLUSTER)
+            .forCluster(helixClusterName).build();
+        Map<String, String> values = helixAdminForMv.getConfig(scope, Collections.singletonList(configName));
+        return values == null ? null : values.get(configName);
+      } catch (Exception e) {
+        return null;
+      }
+    });
+    _helixResourceManager.registerMaterializedViewConsistencyManager(_materializedViewConsistencyManager);
+    _helixResourceManager.getSegmentDeletionManager()
+        .registerMaterializedViewConsistencyManager(_materializedViewConsistencyManager);
+
+    // Initialize segment lifecycle event listeners (registered after MV manager so any
+    // listener that fires immediately on registration sees a fully-wired notify path).
     PinotSegmentLifecycleEventListenerManager.getInstance().init(_helixParticipantManager);
 
     LOGGER.info("Starting task resource manager");
@@ -1153,6 +1186,11 @@ private void stopPinotController() {
       LOGGER.info("Stopping Jersey admin API");
       _adminApp.stop();
 
+      if (_materializedViewConsistencyManager != null) {
+        LOGGER.info("Stopping MV consistency manager");
+        _materializedViewConsistencyManager.stop();
+      }
+
       LOGGER.info("Stopping resource manager");
       _helixResourceManager.stop();