HappenLee commented on code in PR #45926: URL: https://github.com/apache/doris/pull/45926#discussion_r1914688107
########## fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java: ########## @@ -0,0 +1,365 @@ +// 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.doris.dictionary; + +import org.apache.doris.analysis.UserIdentity; +import org.apache.doris.catalog.Env; +import org.apache.doris.common.DdlException; +import org.apache.doris.common.io.Text; +import org.apache.doris.common.io.Writable; +import org.apache.doris.common.util.MasterDaemon; +import org.apache.doris.job.extensions.insert.InsertTask; +import org.apache.doris.nereids.parser.NereidsParser; +import org.apache.doris.nereids.trees.plans.commands.info.CreateDictionaryInfo; +import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoDictionaryCommand; +import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand; +import org.apache.doris.persist.CreateDictionaryPersistInfo; +import org.apache.doris.persist.DropDictionaryPersistInfo; +import org.apache.doris.persist.gson.GsonUtils; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.StmtExecutor; +import org.apache.doris.thrift.TUniqueId; + +import com.google.common.collect.ArrayListMultimap; +import com.google.common.collect.ListMultimap; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import com.google.gson.annotations.SerializedName; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.io.DataInput; +import java.io.DataOutput; +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +/** + * Manager for dictionary operations, including creation, deletion, and data loading. + */ +public class DictionaryManager extends MasterDaemon implements Writable { + private static final Logger LOG = LogManager.getLogger(DictionaryManager.class); + + private static final long jobId = -493209151411825L; + + // Lock for protecting dictionaries map + private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true); + + // Map of database name -> dictionary name -> dictionary + @SerializedName(value = "d") + private Map<String, Map<String, Dictionary>> dictionaries = Maps.newConcurrentMap(); + + // dbname -> tablename -> dictname + @SerializedName(value = "t") + private Map<String, ListMultimap<String, String>> dbTableToDicNames = Maps.newConcurrentMap(); + + @SerializedName(value = "i") + private long uniqueId = 0; + + public DictionaryManager() { + super("Dictionary Manager", 10 * 60 * 1000); // run every 10 minutes + } + + @Override + protected void runAfterCatalogReady() { + // Check and update dictionary data in each cycle + try { + checkAndUpdateDictionaries(); + } catch (Exception e) { + LOG.warn("Failed to check and update dictionaries", e); + } + } + + public void lockRead() { + lock.readLock().lock(); + } + + public void unlockRead() { + lock.readLock().unlock(); + } + + public void lockWrite() { + lock.writeLock().lock(); + } + + public void unlockWrite() { + lock.writeLock().unlock(); + } + + /** + * Create a new dictionary based on the provided info. + * + * @throws Exception + */ + public Dictionary createDictionary(ConnectContext ctx, CreateDictionaryInfo info) throws Exception { + // 1. Check if dictionary already exists + if (hasDictionary(info.getDbName(), info.getDictName())) { + if (info.isIfNotExists()) { + return getDictionary(info.getDbName(), info.getDictName()); + } else { + throw new DdlException( + "Dictionary " + info.getDictName() + " already exists in database " + info.getDbName()); + } + } + + Dictionary dictionary; + lockWrite(); + try { + // Create dictionary object + dictionary = new Dictionary(info, ++uniqueId); + // Add to dictionaries map. no throw here. so schedule below is safe. + Map<String, Dictionary> dbDictionaries = dictionaries.computeIfAbsent(info.getDbName(), + k -> Maps.newConcurrentMap()); + dbDictionaries.put(info.getDictName(), dictionary); + ListMultimap<String, String> tableToDicNames = dbTableToDicNames.computeIfAbsent(info.getDbName(), + k -> ArrayListMultimap.create()); + tableToDicNames.put(info.getSourceTableName(), info.getDictName()); + + // Log the creation operation + Env.getCurrentEnv().getEditLog().logCreateDictionary(dictionary); + } finally { + unlockWrite(); + } + return dictionary; + } + + /** + * Delete a dictionary. + * + * @throws DdlException if the dictionary does not exist + */ + public void dropDictionary(ConnectContext ctx, String dbName, String dictName, boolean ifExists) + throws DdlException { + lockWrite(); + Dictionary dic = null; + try { + Map<String, Dictionary> dbDictionaries = dictionaries.get(dbName); + if (dbDictionaries == null || !dbDictionaries.containsKey(dictName)) { + if (!ifExists) { + throw new DdlException("Dictionary " + dictName + " does not exist in database " + dbName); + } + return; + } + dic = dbDictionaries.remove(dictName); + + // remove mapping from table to dict + dbTableToDicNames.get(dbName).remove(dic.getSourceTableName(), dictName); + + // Log the drop operation + Env.getCurrentEnv().getEditLog().logDropDictionary(dbName, dictName); + } finally { + unlockWrite(); + } + // The data in BE doesn't always have the same situation with FE(think FE crash). But we have periodic report + // so that we can drop unknown dictionary at that time. + scheduleDataUnload(ctx, dic); + } + + /** + * Drop all dictionaries in a table. Used when dropping a table. So maybe no db or table records. + */ + public void dropTableDictionaries(String dbName, String tableName) { + lockWrite(); + List<Dictionary> droppedDictionaries = Lists.newArrayList(); + try { + ListMultimap<String, String> tableToDicNames = dbTableToDicNames.get(dbName); + if (tableToDicNames == null) { // this db has no table with dictionary records. + return; + } + // get all dictionary names of this table + List<String> dictNames = tableToDicNames.removeAll(tableName); + if (dictNames == null) { // this table has no dictionaries. + return; + } + // all this db's dictionaries. tableToDicNames is not null so nameToDics must not be null. + Map<String, Dictionary> nameToDics = dictionaries.get(dbName); + for (String dictName : dictNames) { + Dictionary dictionary = nameToDics.remove(dictName); + droppedDictionaries.add(dictionary); + // Log the drop operation + Env.getCurrentEnv().getEditLog().logDropDictionary(dbName, dictName); + } + } finally { + unlockWrite(); + } + for (Dictionary dictionary : droppedDictionaries) { + scheduleDataUnload(null, dictionary); + } + } + + /** + * Drop all dictionaries in a database. Used when dropping a database. + */ + public void dropDbDictionaries(String dbName) { + lockWrite(); + List<Dictionary> droppedDictionaries = Lists.newArrayList(); + try { + // pop and save item from dictionaries + Map<String, Dictionary> dbDictionaries = dictionaries.remove(dbName); + // Log the drop operation + if (dbDictionaries != null) { + for (Map.Entry<String, Dictionary> entry : dbDictionaries.entrySet()) { + droppedDictionaries.add(entry.getValue()); + Env.getCurrentEnv().getEditLog().logDropDictionary(dbName, entry.getKey()); + } + // also drop all name mapping records. + dbTableToDicNames.remove(dbName); + } + } finally { + unlockWrite(); + } + for (Dictionary dictionary : droppedDictionaries) { + scheduleDataUnload(null, dictionary); + } + } + + /** + * Check if a dictionary exists. + */ + public boolean hasDictionary(String dbName, String dictName) { + lockRead(); + try { + Map<String, Dictionary> dbDictionaries = dictionaries.get(dbName); + return dbDictionaries != null && dbDictionaries.containsKey(dictName); + } finally { + unlockRead(); + } + } + + public Map<String, Dictionary> getDictionaries(String dbName) { + lockRead(); + try { + return dictionaries.computeIfAbsent(dbName, k -> Maps.newConcurrentMap()); + } finally { + unlockRead(); + } + } + + /** + * Get a dictionary. + * + * @throws DdlException if the dictionary does not exist + */ + public Dictionary getDictionary(String dbName, String dictName) throws DdlException { + lockRead(); + try { + Map<String, Dictionary> dbDictionaries = dictionaries.get(dbName); + if (dbDictionaries == null || !dbDictionaries.containsKey(dictName)) { + throw new DdlException("Dictionary " + dictName + " does not exist in database " + dbName); + } + return dbDictionaries.get(dictName); + } finally { + unlockRead(); + } + } + + private void checkAndUpdateDictionaries() { + // TODO: Implement dictionary data check and update logic + } + + /// data load and unload are also used for dictionary data check and update to keep data consistency between BE + /// and FE. So they are not private. + + public void scheduleDataLoad(ConnectContext ctx, Dictionary dictionary) throws Exception { + if (ctx == null) { // for run with scheduler, not by command. + // priv check is done in relative(caller) command. so use ADMIN here is ok. + ctx = InsertTask.makeConnectContext(UserIdentity.ADMIN, dictionary.getDbName()); + } + + // not use rerfresh command's executor to avoid potential problems. + StmtExecutor executor = InsertTask.makeStmtExecutor(ctx); + NereidsParser parser = new NereidsParser(); + InsertIntoTableCommand baseCommand = (InsertIntoTableCommand) parser + .parseSingle("insert into " + dictionary.getDbName() + "." + dictionary.getName() + " select * from " + + dictionary.getDbName() + "." + dictionary.getSourceTableName()); + TUniqueId queryId = InsertTask.generateQueryId(); + baseCommand.setLabelName(Optional.of(jobId + "_" + queryId.toString())); + baseCommand.setJobId(jobId); + + dictionary.setStatus(Dictionary.DictionaryStatus.LOADING); + InsertIntoDictionaryCommand command = new InsertIntoDictionaryCommand(baseCommand, dictionary); + + // run with sync + dictionary.increaseVersion(); Review Comment: check here should need protect by lock? -- 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-unsubscr...@doris.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org For additional commands, e-mail: commits-h...@doris.apache.org
