alluxio-master-2.x/table/server/common/src/main/java/alluxio/table/common/udb/PathTranslator.java
/*
 * 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.table.common.udb;

import alluxio.AlluxioURI;
import alluxio.conf.Configuration;
import alluxio.exception.InvalidPathException;
import alluxio.util.ConfigurationUtils;
import alluxio.util.io.PathUtils;

import com.google.common.collect.BiMap;
import com.google.common.collect.HashBiMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * Utilities to convert to and from ufs paths and alluxio paths.
 */
public class PathTranslator {
  private static final Logger LOG = LoggerFactory.getLogger(PathTranslator.class);
  private static final String SCHEME_AUTHORITY_PREFIX =
      ConfigurationUtils.getSchemeAuthority(Configuration.global());
  private static final AlluxioURI BASE_URI = new AlluxioURI(SCHEME_AUTHORITY_PREFIX);
  private final BiMap<AlluxioURI, AlluxioURI> mPathMap;

  /**
   * Construct a path translator.
   */
  public PathTranslator() {
    mPathMap = HashBiMap.create();
  }

  /**
   * Add a mapping to the path translator.
   *
   * @param alluxioPath the alluxio path
   * @param ufsPath the corresponding ufs path
   *
   * @return PathTranslator object
   */
  public PathTranslator addMapping(String alluxioPath, String ufsPath) {
    mPathMap.put(new AlluxioURI(alluxioPath), new AlluxioURI(ufsPath));
    return this;
  }

  /**
   * Returns the corresponding alluxio path, for the specified ufs path.
   *
   * @param ufsPath the ufs path to translate
   * @return the corresponding alluxio path
   * @throws IOException if the ufs path is not mounted
   */
  public String toAlluxioPath(String ufsPath) throws IOException {
    String suffix = ufsPath.endsWith("/") ? "/" : "";
    AlluxioURI ufsUri = new AlluxioURI(ufsPath);
    // first look for an exact match
    if (mPathMap.inverse().containsKey(ufsUri)) {
      AlluxioURI match = mPathMap.inverse().get(ufsUri);
      if (match.equals(ufsUri)) {
        // bypassed UFS path, return as is
        return ufsPath;
      }
      return checkAndAddSchemeAuthority(mPathMap.inverse().get(ufsUri)) + suffix;
    }
    // otherwise match by longest prefix
    BiMap.Entry<AlluxioURI, AlluxioURI> longestPrefix = null;
    int longestPrefixDepth = -1;
    for (BiMap.Entry<AlluxioURI, AlluxioURI> entry : mPathMap.entrySet()) {
      try {
        AlluxioURI valueUri = entry.getValue();
        if (valueUri.isAncestorOf(ufsUri) && valueUri.getDepth() > longestPrefixDepth) {
          longestPrefix = entry;
          longestPrefixDepth = valueUri.getDepth();
        }
      } catch (InvalidPathException e) {
        throw new IOException(e);
      }
    }
    if (longestPrefix == null) {
      // TODO(yuzhu): instead of throwing an exception, mount the path?
      throw new IOException(String
          .format("Failed to translate ufs path (%s). Mapping missing from translator", ufsPath));
    }
    if (longestPrefix.getKey().equals(longestPrefix.getValue())) {
      // return ufsPath if set the key and value to be same when bypass path.
      return ufsPath;
    }
    try {
      String difference = PathUtils.subtractPaths(ufsUri.getPath(),
          longestPrefix.getValue().getPath());
      AlluxioURI mappedUri = longestPrefix.getKey().join(difference);
      return checkAndAddSchemeAuthority(mappedUri) + suffix;
    } catch (InvalidPathException e) {
      throw new IOException(e);
    }
  }

  private static AlluxioURI checkAndAddSchemeAuthority(AlluxioURI input) {
    if (!input.hasScheme()) {
      return new AlluxioURI(BASE_URI, input.getPath(), false);
    }
    return input;
  }
}
