Skip to content
Permalink
Browse files
ATLAS-4556: Add Ozone 'ofs' scheme support for external tables create… 
…d through hive.

Signed-off-by: Pinal Shah <pinal.shah@freestoneinfotech.com>
  • Loading branch information
@mandarambawane @pinal-shah
mandarambawane authored and pinal-shah committed Feb 23, 2022
1 parent 05b27be commit e1b832cba15f81fe6544c642285f89fc8d9abcfd
Showing 2 changed files with 56 additions and 18 deletions.
52 common/src/main/java/org/apache/atlas/utils/AtlasPathExtractorUtil.java
View file
@@ -29,6 +29,7 @@
import org.slf4j.LoggerFactory;

import java.net.URI;
import java.util.stream.IntStream;

public class AtlasPathExtractorUtil {
private static final Logger LOG = LoggerFactory.getLogger(AtlasPathExtractorUtil.class);
@@ -81,6 +82,7 @@ public class AtlasPathExtractorUtil {
public static final String ATTRIBUTE_VOLUME = "volume";
public static final String RELATIONSHIP_OZONE_VOLUME_BUCKET = "ozone_volume_buckets";
public static final String RELATIONSHIP_OZONE_PARENT_CHILDREN = "ozone_parent_children";
public static final String OZONE_SCHEME_NAME = "ofs";

//Google Cloud Storage
public static final String GCS_SCHEME = "gs" + SCHEME_SEPARATOR;
@@ -432,12 +434,29 @@ private static AtlasEntity addOzonePathEntity(Path path, AtlasEntityExtInfo extI

String keyQNamePrefix = ozoneScheme + SCHEME_SEPARATOR + path.toUri().getAuthority();

for (String subDirName : dirPath.split(Path.SEPARATOR)) {

String[] subDirNames = dirPath.split(Path.SEPARATOR);
String[] subDirNameArr = subDirNames;
if (ozoneScheme.equals(OZONE_SCHEME_NAME)) {
subDirNames = IntStream.range(3, subDirNameArr.length)
.mapToObj(i -> subDirNameArr[i])
.toArray(String[]::new);
}

boolean volumeBucketAdded = false;
for (String subDirName : subDirNames) {
if (StringUtils.isEmpty(subDirName)) {
continue;
}

String subDirPath = parentPath + subDirName;
String subDirPath;
if (ozoneScheme.equals(OZONE_SCHEME_NAME) && !volumeBucketAdded) {
subDirPath = "%s%s" + Path.SEPARATOR + "%s" + Path.SEPARATOR + "%s";
subDirPath = String.format(subDirPath, parentPath, subDirNameArr[1], subDirNameArr[2], subDirName);
volumeBucketAdded = true;
} else {
subDirPath = parentPath + subDirName;
}
String subDirQualifiedName = keyQNamePrefix + subDirPath + QNAME_SEP_METADATA_NAMESPACE + metadataNamespace;

ret = context.getEntity(subDirQualifiedName);
@@ -606,14 +625,33 @@ private static String getAbfsStorageAccountName(URI uri) {
}

private static String getOzoneVolumeName(Path path) {
String pathAuthority = path.toUri().getAuthority();
// pathAuthority: "<bucket_name>.<volume_name>.<ozone.service.id>"
return pathAuthority.split("\\.")[1];
String strPath = path.toString();
String volumeName = StringUtils.EMPTY;
if (strPath.startsWith(OZONE_3_SCHEME)) {
String pathAuthority = path.toUri().getAuthority();
volumeName = pathAuthority.split("\\.")[1];
} else if (strPath.startsWith(OZONE_SCHEME)) {
strPath = strPath.replaceAll(OZONE_SCHEME, StringUtils.EMPTY);
if (strPath.split(Path.SEPARATOR).length >= 2) {
volumeName = strPath.split(Path.SEPARATOR)[1];
}
}
return volumeName;
}

private static String getOzoneBucketName(Path path) {
String pathAuthority = path.toUri().getAuthority();
return pathAuthority.split("\\.")[0];
String strPath = path.toString();
String bucketName = StringUtils.EMPTY;
if (strPath.startsWith(OZONE_3_SCHEME)) {
String pathAuthority = path.toUri().getAuthority();
bucketName = pathAuthority.split("\\.")[0];
} else if (strPath.startsWith(OZONE_SCHEME)) {
strPath = strPath.replaceAll(OZONE_SCHEME, StringUtils.EMPTY);
if (strPath.split(Path.SEPARATOR).length >= 3) {
bucketName = strPath.split(Path.SEPARATOR)[2];
}
}
return bucketName;
}

private static String getQualifiedName(String path, String metadataNamespace) {
22 common/src/test/java/org/apache/atlas/utils/AtlasPathExtractorUtilTest.java
View file
@@ -87,20 +87,20 @@ public class AtlasPathExtractorUtilTest {
@DataProvider(name = "ozonePathProvider")
private Object[][] ozonePathProvider(){
return new Object[][]{
{ new OzoneKeyValidator(OZONE_SCHEME, "bucket1.volume1.ozone1.com/files/file.txt",
"files", "bucket1.volume1.ozone1.com/files",
"file.txt", "bucket1.volume1.ozone1.com/files/file.txt")},
{ new OzoneKeyValidator(OZONE_SCHEME, "ozone1.com/volume1/bucket1/files/file.txt",
"files", "ozone1.com/volume1/bucket1/files",
"file.txt", "ozone1.com/volume1/bucket1/files/file.txt")},

{ new OzoneKeyValidator(OZONE_SCHEME, "bucket1.volume1.ozone1:1234/file21.txt",
"file21.txt", "bucket1.volume1.ozone1:1234/file21.txt") },
{ new OzoneKeyValidator(OZONE_SCHEME, "ozone1:1234/volume1/bucket1/file21.txt",
"file21.txt", "ozone1:1234/volume1/bucket1/file21.txt")},

{ new OzoneKeyValidator(OZONE_SCHEME, "bucket1.volume1.ozone1/quarter_one/sales",
"quarter_one", "bucket1.volume1.ozone1/quarter_one",
"sales", "bucket1.volume1.ozone1/quarter_one/sales") },
{ new OzoneKeyValidator(OZONE_SCHEME, "ozone1/volume1/bucket1/quarter_one/sales",
"quarter_one", "ozone1/volume1/bucket1/quarter_one",
"sales", "ozone1/volume1/bucket1/quarter_one/sales")},

{ new OzoneKeyValidator(OZONE_SCHEME, "bucket1.volume1.ozone1/quarter_one/sales/",
"quarter_one", "bucket1.volume1.ozone1/quarter_one",
"sales", "bucket1.volume1.ozone1/quarter_one/sales") },
{ new OzoneKeyValidator(OZONE_SCHEME, "ozone1/volume1/bucket1/quarter_one/sales/",
"quarter_one", "ozone1/volume1/bucket1/quarter_one",
"sales", "ozone1/volume1/bucket1/quarter_one/sales")},

{ new OzoneKeyValidator(OZONE_3_SCHEME, "bucket1.volume1.ozone1/files/file.txt",
"files", "bucket1.volume1.ozone1/files",

0 comments on commit e1b832c

Please sign in to comment.