Jackie-Jiang commented on code in PR #17291: URL: https://github.com/apache/pinot/pull/17291#discussion_r2625181560
########## pinot-common/src/main/java/org/apache/pinot/common/systemtable/SystemTableRegistry.java: ########## @@ -0,0 +1,129 @@ +/** + * 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.common.systemtable; + +import java.util.Collection; +import java.util.Collections; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.ServiceLoader; +import java.util.concurrent.ConcurrentHashMap; +import javax.annotation.Nullable; +import org.apache.helix.HelixAdmin; +import org.apache.pinot.common.config.provider.TableCache; +import org.apache.pinot.spi.systemtable.SystemTable; +import org.apache.pinot.spi.systemtable.SystemTableProvider; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Registry to hold and lifecycle-manage system table providers. + */ +public final class SystemTableRegistry implements AutoCloseable { + public static final SystemTableRegistry INSTANCE = new SystemTableRegistry(); + private static final Logger LOGGER = LoggerFactory.getLogger(SystemTableRegistry.class); + + private final Map<String, SystemTableProvider> _providers = new ConcurrentHashMap<>(); + + public void register(SystemTableProvider provider) { + _providers.put(normalize(provider.getTableName()), provider); + } + + public @Nullable SystemTableProvider get(String tableName) { + return _providers.get(normalize(tableName)); + } + + public boolean isRegistered(String tableName) { + return _providers.containsKey(normalize(tableName)); + } + + public Collection<SystemTableProvider> getProviders() { + return Collections.unmodifiableCollection(_providers.values()); + } + + /** + * Discover and register providers marked with {@link SystemTable} using the available dependencies. + */ + public void registerAnnotatedProviders(@Nullable TableCache tableCache, @Nullable HelixAdmin helixAdmin, + @Nullable String clusterName) { + for (SystemTableProvider discovered : ServiceLoader.load(SystemTableProvider.class)) { + Class<? extends SystemTableProvider> clazz = discovered.getClass(); + if (!clazz.isAnnotationPresent(SystemTable.class)) { + continue; + } + Optional<SystemTableProvider> instantiated = + instantiateProvider(clazz, tableCache, helixAdmin, clusterName); + instantiated.ifPresent(provider -> { + if (isRegistered(provider.getTableName())) { + return; + } + LOGGER.info("Registering system table provider: {}", provider.getTableName()); + register(provider); + }); + } + } + + @Override + public void close() + throws Exception { + for (SystemTableProvider provider : _providers.values()) { + provider.close(); + } + } + + private static String normalize(String tableName) { + return tableName.toLowerCase(Locale.ROOT); + } + + private Optional<SystemTableProvider> instantiateProvider(Class<? extends SystemTableProvider> clazz, + @Nullable TableCache tableCache, @Nullable HelixAdmin helixAdmin, @Nullable String clusterName) { Review Comment: Will they ever be `null`? ########## pinot-common/src/main/java/org/apache/pinot/common/systemtable/SystemTableRegistry.java: ########## @@ -0,0 +1,129 @@ +/** + * 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.common.systemtable; + +import java.util.Collection; +import java.util.Collections; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.ServiceLoader; +import java.util.concurrent.ConcurrentHashMap; +import javax.annotation.Nullable; +import org.apache.helix.HelixAdmin; +import org.apache.pinot.common.config.provider.TableCache; +import org.apache.pinot.spi.systemtable.SystemTable; +import org.apache.pinot.spi.systemtable.SystemTableProvider; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Registry to hold and lifecycle-manage system table providers. + */ +public final class SystemTableRegistry implements AutoCloseable { + public static final SystemTableRegistry INSTANCE = new SystemTableRegistry(); + private static final Logger LOGGER = LoggerFactory.getLogger(SystemTableRegistry.class); + + private final Map<String, SystemTableProvider> _providers = new ConcurrentHashMap<>(); Review Comment: Regular map should be enough as we are not dynamically registering ########## pinot-common/src/main/java/org/apache/pinot/common/systemtable/SystemTableRegistry.java: ########## @@ -0,0 +1,129 @@ +/** + * 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.common.systemtable; + +import java.util.Collection; +import java.util.Collections; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.ServiceLoader; +import java.util.concurrent.ConcurrentHashMap; +import javax.annotation.Nullable; +import org.apache.helix.HelixAdmin; +import org.apache.pinot.common.config.provider.TableCache; +import org.apache.pinot.spi.systemtable.SystemTable; +import org.apache.pinot.spi.systemtable.SystemTableProvider; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Registry to hold and lifecycle-manage system table providers. + */ +public final class SystemTableRegistry implements AutoCloseable { + public static final SystemTableRegistry INSTANCE = new SystemTableRegistry(); + private static final Logger LOGGER = LoggerFactory.getLogger(SystemTableRegistry.class); + + private final Map<String, SystemTableProvider> _providers = new ConcurrentHashMap<>(); + + public void register(SystemTableProvider provider) { + _providers.put(normalize(provider.getTableName()), provider); + } + + public @Nullable SystemTableProvider get(String tableName) { + return _providers.get(normalize(tableName)); + } + + public boolean isRegistered(String tableName) { + return _providers.containsKey(normalize(tableName)); + } + + public Collection<SystemTableProvider> getProviders() { + return Collections.unmodifiableCollection(_providers.values()); + } + + /** + * Discover and register providers marked with {@link SystemTable} using the available dependencies. + */ + public void registerAnnotatedProviders(@Nullable TableCache tableCache, @Nullable HelixAdmin helixAdmin, + @Nullable String clusterName) { + for (SystemTableProvider discovered : ServiceLoader.load(SystemTableProvider.class)) { + Class<? extends SystemTableProvider> clazz = discovered.getClass(); + if (!clazz.isAnnotationPresent(SystemTable.class)) { + continue; + } + Optional<SystemTableProvider> instantiated = + instantiateProvider(clazz, tableCache, helixAdmin, clusterName); + instantiated.ifPresent(provider -> { + if (isRegistered(provider.getTableName())) { + return; + } + LOGGER.info("Registering system table provider: {}", provider.getTableName()); + register(provider); + }); + } + } + + @Override + public void close() + throws Exception { + for (SystemTableProvider provider : _providers.values()) { + provider.close(); + } + } + + private static String normalize(String tableName) { + return tableName.toLowerCase(Locale.ROOT); + } + + private Optional<SystemTableProvider> instantiateProvider(Class<? extends SystemTableProvider> clazz, + @Nullable TableCache tableCache, @Nullable HelixAdmin helixAdmin, @Nullable String clusterName) { + try { + // Prefer the most specific constructor available. + if (tableCache != null && helixAdmin != null && clusterName != null) { + try { + return Optional.of(clazz.getDeclaredConstructor(TableCache.class, HelixAdmin.class, String.class) + .newInstance(tableCache, helixAdmin, clusterName)); Review Comment: We usually use `init()` method to initialize the plug-ins ########## pinot-spi/src/main/java/org/apache/pinot/spi/systemtable/SystemTableRequest.java: ########## @@ -0,0 +1,63 @@ +/** + * 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.spi.systemtable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import javax.annotation.Nullable; + + +/** + * Request sent to a {@link SystemTableProvider} from the system query engine. + */ +public final class SystemTableRequest { Review Comment: Is this enough to cover all system table requests? Why not using the existing request format? ########## pinot-common/src/main/java/org/apache/pinot/common/systemtable/SystemTableRegistry.java: ########## @@ -0,0 +1,129 @@ +/** + * 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.common.systemtable; + +import java.util.Collection; +import java.util.Collections; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.ServiceLoader; +import java.util.concurrent.ConcurrentHashMap; +import javax.annotation.Nullable; +import org.apache.helix.HelixAdmin; +import org.apache.pinot.common.config.provider.TableCache; +import org.apache.pinot.spi.systemtable.SystemTable; +import org.apache.pinot.spi.systemtable.SystemTableProvider; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Registry to hold and lifecycle-manage system table providers. + */ +public final class SystemTableRegistry implements AutoCloseable { + public static final SystemTableRegistry INSTANCE = new SystemTableRegistry(); + private static final Logger LOGGER = LoggerFactory.getLogger(SystemTableRegistry.class); + + private final Map<String, SystemTableProvider> _providers = new ConcurrentHashMap<>(); + + public void register(SystemTableProvider provider) { + _providers.put(normalize(provider.getTableName()), provider); + } + + public @Nullable SystemTableProvider get(String tableName) { Review Comment: (nit) Put `@Nullable` in a separate line ########## pinot-common/src/main/java/org/apache/pinot/common/systemtable/SystemTableRegistry.java: ########## @@ -0,0 +1,129 @@ +/** + * 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.common.systemtable; + +import java.util.Collection; +import java.util.Collections; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.ServiceLoader; +import java.util.concurrent.ConcurrentHashMap; +import javax.annotation.Nullable; +import org.apache.helix.HelixAdmin; +import org.apache.pinot.common.config.provider.TableCache; +import org.apache.pinot.spi.systemtable.SystemTable; +import org.apache.pinot.spi.systemtable.SystemTableProvider; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Registry to hold and lifecycle-manage system table providers. + */ +public final class SystemTableRegistry implements AutoCloseable { + public static final SystemTableRegistry INSTANCE = new SystemTableRegistry(); Review Comment: Directly exposing singleton instance might not be good practice. Given this is just a registry, making it a util class might be good enough (similar to `ConnectorRegistry`) ########## pinot-common/src/main/java/org/apache/pinot/common/systemtable/SystemTableRegistry.java: ########## @@ -0,0 +1,129 @@ +/** + * 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.common.systemtable; + +import java.util.Collection; +import java.util.Collections; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.ServiceLoader; +import java.util.concurrent.ConcurrentHashMap; +import javax.annotation.Nullable; +import org.apache.helix.HelixAdmin; +import org.apache.pinot.common.config.provider.TableCache; +import org.apache.pinot.spi.systemtable.SystemTable; +import org.apache.pinot.spi.systemtable.SystemTableProvider; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Registry to hold and lifecycle-manage system table providers. + */ +public final class SystemTableRegistry implements AutoCloseable { + public static final SystemTableRegistry INSTANCE = new SystemTableRegistry(); + private static final Logger LOGGER = LoggerFactory.getLogger(SystemTableRegistry.class); + + private final Map<String, SystemTableProvider> _providers = new ConcurrentHashMap<>(); + + public void register(SystemTableProvider provider) { + _providers.put(normalize(provider.getTableName()), provider); + } + + public @Nullable SystemTableProvider get(String tableName) { + return _providers.get(normalize(tableName)); + } + + public boolean isRegistered(String tableName) { + return _providers.containsKey(normalize(tableName)); + } + + public Collection<SystemTableProvider> getProviders() { + return Collections.unmodifiableCollection(_providers.values()); + } + + /** + * Discover and register providers marked with {@link SystemTable} using the available dependencies. + */ + public void registerAnnotatedProviders(@Nullable TableCache tableCache, @Nullable HelixAdmin helixAdmin, + @Nullable String clusterName) { + for (SystemTableProvider discovered : ServiceLoader.load(SystemTableProvider.class)) { + Class<? extends SystemTableProvider> clazz = discovered.getClass(); + if (!clazz.isAnnotationPresent(SystemTable.class)) { + continue; + } + Optional<SystemTableProvider> instantiated = + instantiateProvider(clazz, tableCache, helixAdmin, clusterName); + instantiated.ifPresent(provider -> { + if (isRegistered(provider.getTableName())) { + return; + } + LOGGER.info("Registering system table provider: {}", provider.getTableName()); + register(provider); + }); + } + } + + @Override + public void close() + throws Exception { + for (SystemTableProvider provider : _providers.values()) { + provider.close(); + } + } + + private static String normalize(String tableName) { + return tableName.toLowerCase(Locale.ROOT); + } + + private Optional<SystemTableProvider> instantiateProvider(Class<? extends SystemTableProvider> clazz, Review Comment: We are mixing usage of `null` and `Optional`. Suggest make it return `@Nullable` instead ########## pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java: ########## @@ -176,13 +184,15 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ protected BlockingQueue<Pair<String, String>> _multistageCompileQueryQueue; protected ImplicitHybridTableRouteProvider _implicitHybridTableRouteProvider; protected LogicalTableRouteProvider _logicalTableRouteProvider; + protected final SystemTableRegistry _systemTableRegistry; Review Comment: Consider adding a broker request handler for system table. We shouldn't classify it under SSE ########## pinot-common/src/main/java/org/apache/pinot/common/systemtable/SystemTableRegistry.java: ########## @@ -0,0 +1,129 @@ +/** + * 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.common.systemtable; + +import java.util.Collection; +import java.util.Collections; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.ServiceLoader; +import java.util.concurrent.ConcurrentHashMap; +import javax.annotation.Nullable; +import org.apache.helix.HelixAdmin; +import org.apache.pinot.common.config.provider.TableCache; +import org.apache.pinot.spi.systemtable.SystemTable; +import org.apache.pinot.spi.systemtable.SystemTableProvider; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Registry to hold and lifecycle-manage system table providers. + */ +public final class SystemTableRegistry implements AutoCloseable { + public static final SystemTableRegistry INSTANCE = new SystemTableRegistry(); + private static final Logger LOGGER = LoggerFactory.getLogger(SystemTableRegistry.class); + + private final Map<String, SystemTableProvider> _providers = new ConcurrentHashMap<>(); + + public void register(SystemTableProvider provider) { + _providers.put(normalize(provider.getTableName()), provider); + } + + public @Nullable SystemTableProvider get(String tableName) { + return _providers.get(normalize(tableName)); + } + + public boolean isRegistered(String tableName) { + return _providers.containsKey(normalize(tableName)); + } + + public Collection<SystemTableProvider> getProviders() { Review Comment: (minor) Not used ########## pinot-spi/src/main/java/org/apache/pinot/spi/systemtable/SystemTableResponse.java: ########## @@ -0,0 +1,59 @@ +/** + * 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.spi.systemtable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import org.apache.pinot.spi.data.readers.GenericRow; + + +/** + * Response returned by {@link SystemTableProvider} containing materialized rows. + */ +public final class SystemTableResponse { Review Comment: Any specific reason why not reuse the existing response format? Is there certain standard for system table response? -- 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]
