content: /*
   * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
   * (the "License"). You may not use this work except in compliance with the License, which is
   * available at www.apache.org/licenses/LICENSE-2.0
   *
   * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
   * either express or implied, as more fully set forth in the License.
   *
   * See the NOTICE file distributed with this work for information regarding copyright ownership.
   */
  
  package alluxio.master.table;
  
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.exception.ExceptionMessage;
  import alluxio.exception.status.NotFoundException;
  import alluxio.grpc.table.FileStatistics;
  import alluxio.grpc.table.Schema;
  import alluxio.grpc.table.SyncStatus;
  import alluxio.master.journal.JournalContext;
  import alluxio.master.journal.Journaled;
  import alluxio.master.journal.checkpoint.CheckpointName;
  import alluxio.proto.journal.Journal;
  import alluxio.resource.CloseableIterator;
  import alluxio.table.common.udb.UdbContext;
  import alluxio.table.common.udb.UdbTable;
  import alluxio.table.common.udb.UnderDatabase;
  import alluxio.util.CommonUtils;
  import alluxio.util.ConfigurationUtils;
  import alluxio.util.executor.ExecutorServiceFactories;
  
  import com.fasterxml.jackson.core.JsonProcessingException;
  import com.fasterxml.jackson.databind.ObjectMapper;
  import com.google.common.collect.Iterators;
  import com.google.common.collect.Sets;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.File;
  import java.io.FileNotFoundException;
  import java.io.IOException;
  import java.nio.file.Files;
  import java.nio.file.Paths;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.NoSuchElementException;
  import java.util.Set;
  import java.util.concurrent.Callable;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.function.Supplier;
  import javax.annotation.Nullable;
  
  /**
   * The database implementation that manages a collection of tables.
   */
  public class Database implements Journaled {
    private static final Logger LOG = LoggerFactory.getLogger(Database.class);
  
    private final CatalogContext mContext;
    private final String mType;
    private final String mName;
    private final Map<String, Table> mTables;
    private final UnderDatabase mUdb;
    private final CatalogConfiguration mConfig;
    private final Set<String> mIgnoreTables;
    private final String mConfigPath;
    private DbConfig mDbConfig;
    private final long mUdbSyncTimeoutMs =
        Configuration.getMs(PropertyKey.TABLE_CATALOG_UDB_SYNC_TIMEOUT);
  
    private DatabaseInfo mDatabaseInfo;
  
    private Database(CatalogContext context, String type, String name, UnderDatabase udb,
        CatalogConfiguration config) {
      mContext = context;
      mType = type;
      mName = name;
      mTables = new ConcurrentHashMap<>();
      mUdb = udb;
      mConfig = config;
      mIgnoreTables = Sets.newHashSet(
          ConfigurationUtils.parseAsList(mConfig.get(CatalogProperty.DB_IGNORE_TABLES), ","));
      mConfigPath = mConfig.get(CatalogProperty.DB_CONFIG_FILE);
      mDbConfig = DbConfig.empty();
    }
  
    /**
     * Creates an instance of a database.
     *
     * @param catalogContext the catalog context
     * @param udbContext the db context
     * @param type the database type
     * @param name the database name
     * @param configMap the configuration
     * @return the database instance
     */
    public static Database create(CatalogContext catalogContext, UdbContext udbContext, String type,
        String name, Map<String, String> configMap) {
      CatalogConfiguration configuration = new CatalogConfiguration(configMap);
      try {
        UnderDatabase udb = udbContext.getUdbRegistry()
            .create(udbContext, type, configuration.getUdbConfiguration(type));
        return new Database(catalogContext, type, name, udb, configuration);
      } catch (Exception e) {
        throw new IllegalArgumentException("Creating udb failed for database name: " + name, e);
      }
    }
  
    /**
     * @return the catalog context
     */
    public CatalogContext getContext() {
      return mContext;
    }
  
    /**
     * @return returns the database name
     */
    public String getName() {
      return mName;
    }
  
    /**
     * @return returns database info
     */
    public DatabaseInfo getDatabaseInfo() {
      return mDatabaseInfo;
    }
  
    /**
     * @return returns the database type
     */
    public String getType() {
      return mType;
    }
  
    /**
     * @return the {@link UnderDatabase}
     */
    public UnderDatabase getUdb() {
      return mUdb;
    }
  
    /**
     * @return the list of all tables
     */
    public List<Table> getTables() {
      return new ArrayList<>(mTables.values());
    }
  
    /**
     * @param tableName the table name
     * @return the {@link Table} for the specified table name
     */
    public Table getTable(String tableName) throws NotFoundException {
      Table table = mTables.get(tableName);
      if (table == null) {
        throw new NotFoundException(ExceptionMessage.TABLE_DOES_NOT_EXIST
            .getMessage(tableName, mName));
      }
      return table;
    }
  
    /**
     * Creates a new table within this database.
     *
     * @param tableName the new table name
     * @param schema the schema for the table
     * @return the {@link Table} for the newly created table
     */
    public Table createTable(String tableName, Schema schema) {
      // TODO(gpang): implement
      return mTables.get(tableName);
    }
  
    /**
     * @param tableName the table name
     * @return statistics for the specified table name
     */
    public Map<String, FileStatistics> getStatistics(String tableName) {
      // TODO(gpang): implement
      return Collections.emptyMap();
    }
  
    /**
     *
     * @return the configuration for the database
     */
    public Map<String, String> getConfig() {
      return mConfig.getMap();
    }
  
    /**
     * Syncs the metadata from the under db. To avoid concurrent sync operations, this requires
     * external synchronization.
     *
     * @param context journal context
     * @return the resulting sync status
     */
    public SyncStatus sync(JournalContext context) throws IOException {
      // Keep track of the status of each syncing table.
      // Synchronization is necessary if accessed concurrently from multiple threads
      SyncStatus.Builder builder = SyncStatus.newBuilder();
  
      if (!mConfigPath.equals(CatalogProperty.DB_CONFIG_FILE.getDefaultValue())) {
        if (!Files.exists(Paths.get(mConfigPath))) {
          throw new FileNotFoundException(mConfigPath);
        }
        ObjectMapper mapper = new ObjectMapper();
        try {
          mDbConfig = mapper.readValue(new File(mConfigPath), DbConfig.class);
        } catch (JsonProcessingException e) {
          LOG.error("Failed to deserialize UDB config file {}, stays unsynced", mConfigPath, e);
          throw e;
        }
      }
      DatabaseInfo newDbInfo = mUdb.getDatabaseInfo();
      if (!newDbInfo.equals(mDatabaseInfo)) {
        applyAndJournal(context, Journal.JournalEntry.newBuilder()
            .setUpdateDatabaseInfo(toJournalProto(newDbInfo, mName)).build());
      }
  
      Set<String> udbTableNames = new HashSet<>(mUdb.getTableNames());
  
      // keeps track of how many tables have been synced
      final AtomicInteger tablesSynced = new AtomicInteger();
      // # of synced tables, after which a log message is printed for progress
      final int progressBatch =
          (udbTableNames.size() < 100) ? udbTableNames.size() : udbTableNames.size() / 10;
  
      // sync each table in parallel, with the executor service
      List<Callable<Void>> tasks = new ArrayList<>(udbTableNames.size());
      final Database thisDb = this;
      for (String tableName : udbTableNames) {
        if (mIgnoreTables.contains(tableName)) {
          // this table should be ignored.
          builder.addTablesIgnored(tableName);
          tablesSynced.incrementAndGet();
          continue;
        }
        tasks.add(() -> {
          // Save all exceptions
          try {
            Table previousTable = mTables.get(tableName);
            UdbTable udbTable = mUdb.getTable(tableName, mDbConfig.getUdbBypassSpec());
            Table newTable = Table.create(thisDb, udbTable, previousTable);
  
            if (newTable != null) {
              // table was created or was updated
              alluxio.proto.journal.Table.AddTableEntry addTableEntry
                  = newTable.getTableJournalProto();
              Journal.JournalEntry entry =
                  Journal.JournalEntry.newBuilder().setAddTable(addTableEntry).build();
              applyAndJournal(context, entry);
              // separate the possible big table entry into multiple smaller table partitions entry
              newTable.getTablePartitionsJournalProto().forEach((partitionsEntry) -> {
                applyAndJournal(context, Journal.JournalEntry
                    .newBuilder().setAddTablePartitions(partitionsEntry).build());
              });
              synchronized (builder) {
                builder.addTablesUpdated(tableName);
              }
            } else {
              synchronized (builder) {
                builder.addTablesUnchanged(tableName);
              }
            }
          } catch (Exception e) {
            LOG.error(String.format("Sync thread failed for %s.%s", thisDb.mName, tableName), e);
            synchronized (builder) {
              builder.putTablesErrors(tableName, e.toString());
            }
          } finally {
            int syncedTables = tablesSynced.incrementAndGet();
            int percentage = -1;
            // Only log at regular intervals, or when complete
            if (syncedTables % progressBatch == 0) {
              // compute percentage, cap at 99%
              percentage = Math.min(Math.round(100.0f * syncedTables / udbTableNames.size()), 99);
            }
            if (syncedTables == udbTableNames.size()) {
              percentage = 100;
            }
            if (percentage != -1) {
              LOG.info("Syncing db {} progress: completed {} of {} tables ({}%)", mName, syncedTables,
                  udbTableNames.size(), percentage);
            }
          }
          return null;
        });
      }
  
      // create a thread pool to parallelize the sync
      int threads;
      try {
        threads = Integer.parseInt(mConfig.get(CatalogProperty.DB_SYNC_THREADS));
      } catch (NumberFormatException e) {
        LOG.warn("Catalog property {} with value {} cannot be parsed as an int",
            CatalogProperty.DB_SYNC_THREADS.getName(), mConfig.get(CatalogProperty.DB_SYNC_THREADS));
        threads = CatalogProperty.DEFAULT_DB_SYNC_THREADS;
      }
      if (threads < 1) {
        // if invalid, set to the default
        threads = CatalogProperty.DEFAULT_DB_SYNC_THREADS;
      }
      ExecutorService service =
          ExecutorServiceFactories.fixedThreadPool(String.format("Catalog-Sync-%s", mName), threads)
              .create();
      try {
        CommonUtils.invokeAll(service, tasks, mUdbSyncTimeoutMs);
      } catch (Exception e) {
        throw new IOException("Failed to sync database " + mName + ". error: " + e.toString(), e);
      } finally {
        // shutdown the thread pool
        service.shutdownNow();
        String errorMessage =
            String.format("waiting for db-sync thread pool to shut down. db: %s", mName);
        try {
          if (!service.awaitTermination(5, TimeUnit.SECONDS)) {
            LOG.warn("Timed out " + errorMessage);
          }
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          LOG.warn("Interrupted while " + errorMessage);
        }
      }
  
      for (Table existingTable : mTables.values()) {
        if (!udbTableNames.contains(existingTable.getName())) {
          // this table no longer exists in udb
          alluxio.proto.journal.Table.RemoveTableEntry removeTableEntry =
              alluxio.proto.journal.Table.RemoveTableEntry.newBuilder()
                  .setDbName(mName)
                  .setTableName(existingTable.getName())
                  .setVersion(existingTable.getVersion())
                  .build();
          Journal.JournalEntry entry = Journal.JournalEntry.newBuilder()
              .setRemoveTable(removeTableEntry)
              .build();
          applyAndJournal(context, entry);
          builder.addTablesRemoved(existingTable.getName());
        }
      }
      return builder.build();
    }
  
    @Override
    public void applyAndJournal(Supplier<JournalContext> context, Journal.JournalEntry entry) {
      // This is journaled differently from others components, since optimistic concurrency control
      // is utilized. There are no external locks for the table, so the locking will happen during
      // the access of the tables map.
      processJournalEntryInternal(entry, context.get());
    }
  
    @Override
    public boolean processJournalEntry(Journal.JournalEntry entry) {
      // Do not journal when processing journal entries
      return processJournalEntryInternal(entry, null);
    }
  
    /**
     * @param entry the journal entry to process
     * @param context the journal context, will not journal if null
     * @return whether the entry type is supported by this journaled object
     */
    private boolean processJournalEntryInternal(Journal.JournalEntry entry,
        @Nullable JournalContext context) {
      if (entry.hasAddTable()) {
        return applyAddTable(context, entry);
      }
      if (entry.hasAddTablePartitions()) {
        return applyAddTablePartitions(context, entry);
      }
      if (entry.hasRemoveTable()) {
        return applyRemoveTable(context, entry);
      }
      if (entry.hasUpdateDatabaseInfo()) {
        return applyUpdateDbInfo(context, entry);
      }
      return false;
    }
  
    private boolean applyUpdateDbInfo(@Nullable JournalContext context, Journal.JournalEntry entry) {
      alluxio.proto.journal.Table.UpdateDatabaseInfoEntry updateDb = entry.getUpdateDatabaseInfo();
      if (!updateDb.getDbName().equals(mName)) {
        return false;
      }
      if (context != null) {
        context.append(entry);
      }
      mDatabaseInfo = new DatabaseInfo(updateDb.getLocation(), updateDb.getOwnerName(),
          updateDb.getOwnerType(), updateDb.getComment(), updateDb.getParameterMap());
      return true;
    }
  
    private boolean applyAddTable(@Nullable JournalContext context, Journal.JournalEntry entry) {
      alluxio.proto.journal.Table.AddTableEntry addTable = entry.getAddTable();
      if (!addTable.getDbName().equals(mName)) {
        return false;
      }
  
      Table newTable = Table.create(this, addTable);
      mTables.compute(newTable.getName(), (key, existingTable) -> {
        boolean writeNewTable = false;
        if (existingTable == null && (newTable.getVersion() == Table.FIRST_VERSION)) {
          // this table is being newly inserted, and has the expected first version
          LOG.info("Adding new table {}.{}", mName, newTable.getName());
          writeNewTable = true;
        }
  
        if (existingTable != null && (newTable.getPreviousVersion() == existingTable.getVersion())) {
          // Previous table already exists, and matches the new table's previous version
          LOG.info("Updating table {}.{} to version {}", mName, newTable.getName(),
              newTable.getVersion());
          writeNewTable = true;
        }
  
        if (writeNewTable) {
          // The new table has been successfully validated, so update the map with the new table,
          // and journal the entry if the journal context exists.
          if (context != null) {
            context.append(entry);
          }
          return newTable;
        } else {
          // The table to add does not validate with the existing table, so another thread must
          // have updated the map. Do not modify the map.
          return existingTable;
        }
      });
  
      return true;
    }
  
    private boolean applyAddTablePartitions(@Nullable JournalContext context,
        Journal.JournalEntry entry) {
      alluxio.proto.journal.Table.AddTablePartitionsEntry addTablePartitions
          = entry.getAddTablePartitions();
      if (!addTablePartitions.getDbName().equals(mName)) {
        return false;
      }
  
      mTables.compute(addTablePartitions.getTableName(), (key, existingTable) -> {
        if (existingTable != null) {
          if (addTablePartitions.getVersion() == existingTable.getVersion()) {
            LOG.info("Adding {} partitions to table {}.{}", addTablePartitions.getPartitionsCount(),
                mName, addTablePartitions.getTableName());
            if (context != null) {
              context.append(entry);
            }
            existingTable.addPartitions(addTablePartitions);
            return existingTable;
          }
          LOG.info("Will not add partitions to table {}.{}, because of mismatched versions. "
                  + "version-to-add-partitions: {} existing-version: {}",
              mName, addTablePartitions.getTableName(),
              addTablePartitions.getVersion(), existingTable.getVersion());
        }
        LOG.debug("Cannot add partitions to table {}.{}, because it does not exist.", mName,
            addTablePartitions.getTableName());
        return existingTable;
      });
  
      return true;
    }
  
    private boolean applyRemoveTable(@Nullable JournalContext context, Journal.JournalEntry entry) {
      alluxio.proto.journal.Table.RemoveTableEntry removeTable = entry.getRemoveTable();
      if (!removeTable.getDbName().equals(mName)) {
        return false;
      }
  
      mTables.compute(removeTable.getTableName(), (key, existingTable) -> {
        if (existingTable != null) {
          if (removeTable.getVersion() == existingTable.getVersion()) {
            // this table is being removed, and has the expected version
            LOG.info("Removing table {}.{}", mName, removeTable.getTableName());
            if (context != null) {
              context.append(entry);
            }
            return null;
          }
          LOG.info("Will not remove table {}.{}, because of mismatched versions. "
                  + "version-to-delete: {} existing-version: {}", mName, removeTable.getTableName(),
              removeTable.getVersion(), existingTable.getVersion());
        }
        LOG.debug("Cannot remove table {}.{}, because it does not exist.", mName,
            removeTable.getTableName());
        return existingTable;
      });
  
      return true;
    }
  
    @Override
    public void resetState() {
      mTables.clear();
    }
  
    private Iterator<Journal.JournalEntry> getTableIterator() {
      final Iterator<Table> it = getTables().iterator();
      return new Iterator<Journal.JournalEntry>() {
        private Table mEntry = null;
        private Iterator<alluxio.proto.journal.Table.AddTablePartitionsEntry> mPartitionIterator;
  
        @Override
        public boolean hasNext() {
          if (mEntry != null) {
            return true;
          }
          if (mPartitionIterator != null && mPartitionIterator.hasNext()) {
            return true;
          }
          if (it.hasNext()) {
            mEntry = it.next();
            mPartitionIterator = mEntry.getTablePartitionsJournalProto().iterator();
            return true;
          }
          return false;
        }
  
        @Override
        public Journal.JournalEntry next() {
          if (!hasNext()) {
            throw new NoSuchElementException();
          }
          if (mEntry != null) {
            Table table = mEntry;
            mEntry = null;
            alluxio.proto.journal.Table.AddTableEntry addTableEntry = table.getTableJournalProto();
            return Journal.JournalEntry.newBuilder().setAddTable(addTableEntry).build();
          }
          if (mPartitionIterator != null && mPartitionIterator.hasNext()) {
            return Journal.JournalEntry.newBuilder()
                .setAddTablePartitions(mPartitionIterator.next()).build();
          }
          // should not reach here
          throw new NoSuchElementException();
        }
  
        @Override
        public void remove() {
          throw new UnsupportedOperationException(
              "GetTableIterator#Iterator#remove is not supported.");
        }
      };
    }
  
    @Override
    public CloseableIterator<Journal.JournalEntry> getJournalEntryIterator() {
      Journal.JournalEntry entry = Journal.JournalEntry.newBuilder().setUpdateDatabaseInfo(
          toJournalProto(getDatabaseInfo(), mName)).build();
      return CloseableIterator.noopCloseable(
          Iterators.concat(Iterators.singletonIterator(entry), getTableIterator()));
    }
  
    @Override
    public CheckpointName getCheckpointName() {
      return CheckpointName.TABLE_MASTER_DATABASE;
    }
  
    /**
     * @param dbInfo database info object
     * @param dbName database name
     * @return the journal proto representation
     */
    public static alluxio.proto.journal.Table.UpdateDatabaseInfoEntry toJournalProto(
        DatabaseInfo dbInfo, String dbName) {
      alluxio.proto.journal.Table.UpdateDatabaseInfoEntry.Builder builder =
          alluxio.proto.journal.Table.UpdateDatabaseInfoEntry.newBuilder()
              .setDbName(dbName).putAllParameter(dbInfo.getParameters());
      if (dbInfo.getComment() != null) {
        builder.setComment(dbInfo.getComment());
      }
      if (dbInfo.getLocation() != null) {
        builder.setLocation(dbInfo.getLocation());
      }
      if (dbInfo.getOwnerName() != null) {
        builder.setOwnerName(dbInfo.getOwnerName());
      }
      if (dbInfo.getOwnerType() != null) {
        builder.setOwnerType(dbInfo.getOwnerType());
      }
      return builder.build();
    }
  }
  