You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2016/09/19 22:42:19 UTC
[33/34] hive git commit: HIVE-14700 : clean up file/txn information
via a metastore thread (Sergey Shelukhin)
http://git-wip-us.apache.org/repos/asf/hive/blob/70299dc4/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
index 45a80e3..0e3035a 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
@@ -1493,7 +1493,8 @@ public class Driver implements CommandProcessor {
break;
default: return null;
}
- return (t != null && !t.isTemporary() && AcidUtils.isMmTable(t)) ? t : null;
+ return (t != null && !t.isTemporary()
+ && MetaStoreUtils.isMmTable(t.getParameters())) ? t : null;
}
private CommandProcessorResponse rollback(CommandProcessorResponse cpr) {
http://git-wip-us.apache.org/repos/asf/hive/blob/70299dc4/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java b/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
index 70b129e..5705145 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
@@ -36,6 +36,7 @@ import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.metastore.api.DataOperationType;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
+import org.apache.hadoop.hive.metastore.MetaStoreUtils;
import org.apache.hadoop.hive.metastore.TransactionalValidationListener;
import org.apache.hadoop.hive.ql.ErrorMsg;
import org.apache.hadoop.hive.ql.metadata.Table;
@@ -1082,12 +1083,6 @@ public class AcidUtils {
return tableIsTransactional != null && tableIsTransactional.equalsIgnoreCase("true");
}
- public static boolean isMmTable(Table table) {
- // TODO: perhaps it should be a 3rd value for 'transactional'?
- String value = table.getProperty(hive_metastoreConstants.TABLE_IS_MM);
- return value != null && value.equalsIgnoreCase("true");
- }
-
/**
* Sets the acidOperationalProperties in the configuration object argument.
* @param conf Mutable configuration object
http://git-wip-us.apache.org/repos/asf/hive/blob/70299dc4/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
index f3609df..10d3f0e 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
@@ -1478,7 +1478,7 @@ public class Hive {
boolean hasFollowingStatsTask, Long mmWriteId) throws HiveException {
Table tbl = getTable(tableName);
boolean isMmTableWrite = (mmWriteId != null);
- Preconditions.checkState(isMmTableWrite == AcidUtils.isMmTable(tbl));
+ Preconditions.checkState(isMmTableWrite == MetaStoreUtils.isMmTable(tbl.getParameters()));
loadPartition(loadPath, tbl, partSpec, replace, inheritTableSpecs,
isSkewedStoreAsSubdir, isSrcLocal, isAcid, hasFollowingStatsTask, mmWriteId);
if (isMmTableWrite) {
http://git-wip-us.apache.org/repos/asf/hive/blob/70299dc4/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
index a0ce3a6..fa7c29b 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
@@ -6569,7 +6569,7 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer {
}
boolean isNonNativeTable = dest_tab.isNonNative();
- isMmTable = AcidUtils.isMmTable(dest_tab);
+ isMmTable = MetaStoreUtils.isMmTable(dest_tab.getParameters());
if (isNonNativeTable || isMmTable) {
queryTmpdir = dest_path;
} else {
@@ -6642,7 +6642,7 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer {
dest_path = new Path(tabPath.toUri().getScheme(), tabPath.toUri()
.getAuthority(), partPath.toUri().getPath());
- isMmTable = AcidUtils.isMmTable(dest_tab);
+ isMmTable = MetaStoreUtils.isMmTable(dest_tab.getParameters());
queryTmpdir = isMmTable ? dest_path : ctx.getTempDirForPath(dest_path);
Utilities.LOG14535.info("createFS for partition specifying " + queryTmpdir + " from " + dest_path);
table_desc = Utilities.getTableDesc(dest_tab);
http://git-wip-us.apache.org/repos/asf/hive/blob/70299dc4/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorThread.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorThread.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorThread.java
index 4d6e24e..92d9f28 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorThread.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorThread.java
@@ -65,7 +65,6 @@ abstract class CompactorThread extends Thread implements MetaStoreThread {
@Override
public void setThreadId(int threadId) {
this.threadId = threadId;
-
}
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/70299dc4/ql/src/test/org/apache/hadoop/hive/ql/io/TestAcidUtils.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/io/TestAcidUtils.java b/ql/src/test/org/apache/hadoop/hive/ql/io/TestAcidUtils.java
index a7ff9a3..0d177be 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/io/TestAcidUtils.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/io/TestAcidUtils.java
@@ -33,10 +33,10 @@ import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
import org.apache.hadoop.hive.ql.io.AcidUtils.AcidOperationalProperties;
import org.apache.hadoop.hive.ql.io.orc.TestInputOutputFormat;
-import org.apache.hadoop.hive.ql.io.orc.TestInputOutputFormat.MockFile;
-import org.apache.hadoop.hive.ql.io.orc.TestInputOutputFormat.MockFileSystem;
-import org.apache.hadoop.hive.ql.io.orc.TestInputOutputFormat.MockPath;
import org.apache.hadoop.hive.shims.HadoopShims.HdfsFileStatusWithId;
+import org.apache.hive.common.util.MockFileSystem;
+import org.apache.hive.common.util.MockFileSystem.MockFile;
+import org.apache.hive.common.util.MockFileSystem.MockPath;
import org.junit.Assert;
import org.junit.Test;
@@ -179,7 +179,7 @@ public class TestAcidUtils {
new MockFile("mock:/tbl/part1/delta_050_100/bucket_0", 0, new byte[0]),
new MockFile("mock:/tbl/part1/delta_101_101/bucket_0", 0, new byte[0]));
AcidUtils.Directory dir =
- AcidUtils.getAcidState(new TestInputOutputFormat.MockPath(fs,
+ AcidUtils.getAcidState(new MockPath(fs,
"mock:/tbl/part1"), conf, new ValidReadTxnList("100:" + Long.MAX_VALUE + ":"));
assertEquals(null, dir.getBaseDirectory());
List<FileStatus> obsolete = dir.getObsolete();
@@ -221,7 +221,7 @@ public class TestAcidUtils {
new MockFile("mock:/tbl/part1/delta_050_105/bucket_0", 0, new byte[0]),
new MockFile("mock:/tbl/part1/delta_90_120/bucket_0", 0, new byte[0]));
AcidUtils.Directory dir =
- AcidUtils.getAcidState(new TestInputOutputFormat.MockPath(fs,
+ AcidUtils.getAcidState(new MockPath(fs,
"mock:/tbl/part1"), conf, new ValidReadTxnList("100:" + Long.MAX_VALUE + ":"));
assertEquals("mock:/tbl/part1/base_49", dir.getBaseDirectory().toString());
List<FileStatus> obsolete = dir.getObsolete();
@@ -517,7 +517,7 @@ public class TestAcidUtils {
new MockFile("mock:/tbl/part1/delete_delta_050_105/bucket_0", 0, new byte[0]),
new MockFile("mock:/tbl/part1/delete_delta_110_110/bucket_0", 0, new byte[0]));
AcidUtils.Directory dir =
- AcidUtils.getAcidState(new TestInputOutputFormat.MockPath(fs,
+ AcidUtils.getAcidState(new MockPath(fs,
"mock:/tbl/part1"), conf, new ValidReadTxnList("100:" + Long.MAX_VALUE + ":"));
assertEquals("mock:/tbl/part1/base_49", dir.getBaseDirectory().toString());
List<FileStatus> obsolete = dir.getObsolete();
http://git-wip-us.apache.org/repos/asf/hive/blob/70299dc4/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java b/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java
index 2c1bb6f..28a4f9d 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java
@@ -21,10 +21,7 @@ import static org.junit.Assert.*;
import java.io.DataInput;
import java.io.DataOutput;
-import java.io.FileNotFoundException;
import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
import java.security.PrivilegedExceptionAction;
import java.sql.Date;
import java.sql.Timestamp;
@@ -32,27 +29,16 @@ import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
-import java.util.Map;
import java.util.Properties;
-import java.util.Set;
import java.util.TimeZone;
-import java.util.TreeSet;
import org.apache.commons.codec.binary.Base64;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.BlockLocation;
-import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FSInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hive.common.type.HiveDecimal;
import org.apache.hadoop.hive.conf.HiveConf;
@@ -113,7 +99,11 @@ import org.apache.hadoop.mapred.OutputFormat;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.util.Progressable;
+import org.apache.hive.common.util.MockFileSystem;
+import org.apache.hive.common.util.MockFileSystem.MockBlock;
+import org.apache.hive.common.util.MockFileSystem.MockFile;
+import org.apache.hive.common.util.MockFileSystem.MockOutputStream;
+import org.apache.hive.common.util.MockFileSystem.MockPath;
import org.apache.orc.OrcProto;
import org.junit.Before;
import org.junit.Rule;
@@ -921,534 +911,6 @@ public class TestInputOutputFormat {
null, null, true);
}
- public static class MockBlock {
- int offset;
- int length;
- final String[] hosts;
-
- public MockBlock(String... hosts) {
- this.hosts = hosts;
- }
-
- public void setOffset(int offset) {
- this.offset = offset;
- }
-
- public void setLength(int length) {
- this.length = length;
- }
-
- @Override
- public String toString() {
- StringBuilder buffer = new StringBuilder();
- buffer.append("block{offset: ");
- buffer.append(offset);
- buffer.append(", length: ");
- buffer.append(length);
- buffer.append(", hosts: [");
- for(int i=0; i < hosts.length; i++) {
- if (i != 0) {
- buffer.append(", ");
- }
- buffer.append(hosts[i]);
- }
- buffer.append("]}");
- return buffer.toString();
- }
- }
-
- public static class MockFile {
- final Path path;
- int blockSize;
- int length;
- MockBlock[] blocks;
- byte[] content;
-
- public MockFile(String path, int blockSize, byte[] content,
- MockBlock... blocks) {
- this.path = new Path(path);
- this.blockSize = blockSize;
- this.blocks = blocks;
- this.content = content;
- this.length = content.length;
- int offset = 0;
- for(MockBlock block: blocks) {
- block.offset = offset;
- block.length = Math.min(length - offset, blockSize);
- offset += block.length;
- }
- }
-
- @Override
- public int hashCode() {
- return path.hashCode() + 31 * length;
- }
-
- @Override
- public boolean equals(final Object obj) {
- if (!(obj instanceof MockFile)) { return false; }
- return ((MockFile) obj).path.equals(this.path) && ((MockFile) obj).length == this.length;
- }
-
- @Override
- public String toString() {
- StringBuilder buffer = new StringBuilder();
- buffer.append("mockFile{path: ");
- buffer.append(path.toString());
- buffer.append(", blkSize: ");
- buffer.append(blockSize);
- buffer.append(", len: ");
- buffer.append(length);
- buffer.append(", blocks: [");
- for(int i=0; i < blocks.length; i++) {
- if (i != 0) {
- buffer.append(", ");
- }
- buffer.append(blocks[i]);
- }
- buffer.append("]}");
- return buffer.toString();
- }
- }
-
- static class MockInputStream extends FSInputStream {
- final MockFile file;
- int offset = 0;
-
- public MockInputStream(MockFile file) throws IOException {
- this.file = file;
- }
-
- @Override
- public void seek(long offset) throws IOException {
- this.offset = (int) offset;
- }
-
- @Override
- public long getPos() throws IOException {
- return offset;
- }
-
- @Override
- public boolean seekToNewSource(long l) throws IOException {
- return false;
- }
-
- @Override
- public int read() throws IOException {
- if (offset < file.length) {
- return file.content[offset++] & 0xff;
- }
- return -1;
- }
- }
-
- public static class MockPath extends Path {
- private final FileSystem fs;
- public MockPath(FileSystem fs, String path) {
- super(path);
- this.fs = fs;
- }
- @Override
- public FileSystem getFileSystem(Configuration conf) {
- return fs;
- }
- }
-
- public static class MockOutputStream extends FSDataOutputStream {
- private final MockFile file;
-
- public MockOutputStream(MockFile file) throws IOException {
- super(new DataOutputBuffer(), null);
- this.file = file;
- }
-
- /**
- * Set the blocks and their location for the file.
- * Must be called after the stream is closed or the block length will be
- * wrong.
- * @param blocks the list of blocks
- */
- public void setBlocks(MockBlock... blocks) {
- file.blocks = blocks;
- int offset = 0;
- int i = 0;
- while (offset < file.length && i < blocks.length) {
- blocks[i].offset = offset;
- blocks[i].length = Math.min(file.length - offset, file.blockSize);
- offset += blocks[i].length;
- i += 1;
- }
- }
-
- @Override
- public void close() throws IOException {
- super.close();
- DataOutputBuffer buf = (DataOutputBuffer) getWrappedStream();
- file.length = buf.getLength();
- file.content = new byte[file.length];
- MockBlock block = new MockBlock("host1");
- block.setLength(file.length);
- setBlocks(block);
- System.arraycopy(buf.getData(), 0, file.content, 0, file.length);
- }
-
- @Override
- public String toString() {
- return "Out stream to " + file.toString();
- }
- }
-
- public static class MockFileSystem extends FileSystem {
- final List<MockFile> files = new ArrayList<MockFile>();
- final Map<MockFile, FileStatus> fileStatusMap = new HashMap<>();
- Path workingDir = new Path("/");
- // statics for when the mock fs is created via FileSystem.get
- private static String blockedUgi = null;
- private final static List<MockFile> globalFiles = new ArrayList<MockFile>();
- protected Statistics statistics;
-
- public MockFileSystem() {
- // empty
- }
-
- @Override
- public void initialize(URI uri, Configuration conf) {
- setConf(conf);
- statistics = getStatistics("mock", getClass());
- }
-
- public MockFileSystem(Configuration conf, MockFile... files) {
- setConf(conf);
- this.files.addAll(Arrays.asList(files));
- statistics = getStatistics("mock", getClass());
- }
-
- public static void setBlockedUgi(String s) {
- blockedUgi = s;
- }
-
- void clear() {
- files.clear();
- }
-
- @Override
- public URI getUri() {
- try {
- return new URI("mock:///");
- } catch (URISyntaxException err) {
- throw new IllegalArgumentException("huh?", err);
- }
- }
-
- // increments file modification time
- public void touch(MockFile file) {
- if (fileStatusMap.containsKey(file)) {
- FileStatus fileStatus = fileStatusMap.get(file);
- FileStatus fileStatusNew = new FileStatus(fileStatus.getLen(), fileStatus.isDirectory(),
- fileStatus.getReplication(), fileStatus.getBlockSize(),
- fileStatus.getModificationTime() + 1, fileStatus.getAccessTime(),
- fileStatus.getPermission(), fileStatus.getOwner(), fileStatus.getGroup(),
- fileStatus.getPath());
- fileStatusMap.put(file, fileStatusNew);
- }
- }
-
- @SuppressWarnings("serial")
- public static class MockAccessDenied extends IOException {
- }
-
- @Override
- public FSDataInputStream open(Path path, int i) throws IOException {
- statistics.incrementReadOps(1);
- checkAccess();
- MockFile file = findFile(path);
- if (file != null) return new FSDataInputStream(new MockInputStream(file));
- throw new IOException("File not found: " + path);
- }
-
- private MockFile findFile(Path path) {
- for (MockFile file: files) {
- if (file.path.equals(path)) {
- return file;
- }
- }
- for (MockFile file: globalFiles) {
- if (file.path.equals(path)) {
- return file;
- }
- }
- return null;
- }
-
- private void checkAccess() throws IOException {
- if (blockedUgi == null) return;
- if (!blockedUgi.equals(UserGroupInformation.getCurrentUser().getShortUserName())) return;
- throw new MockAccessDenied();
- }
-
- @Override
- public FSDataOutputStream create(Path path, FsPermission fsPermission,
- boolean overwrite, int bufferSize,
- short replication, long blockSize,
- Progressable progressable
- ) throws IOException {
- statistics.incrementWriteOps(1);
- checkAccess();
- MockFile file = findFile(path);
- if (file == null) {
- file = new MockFile(path.toString(), (int) blockSize, new byte[0]);
- files.add(file);
- }
- return new MockOutputStream(file);
- }
-
- @Override
- public FSDataOutputStream append(Path path, int bufferSize,
- Progressable progressable
- ) throws IOException {
- statistics.incrementWriteOps(1);
- checkAccess();
- return create(path, FsPermission.getDefault(), true, bufferSize,
- (short) 3, 256 * 1024, progressable);
- }
-
- @Override
- public boolean rename(Path path, Path path2) throws IOException {
- statistics.incrementWriteOps(1);
- checkAccess();
- return false;
- }
-
- @Override
- public boolean delete(Path path) throws IOException {
- statistics.incrementWriteOps(1);
- checkAccess();
- return false;
- }
-
- @Override
- public boolean delete(Path path, boolean b) throws IOException {
- statistics.incrementWriteOps(1);
- checkAccess();
- return false;
- }
-
- @Override
- public RemoteIterator<LocatedFileStatus> listLocatedStatus(final Path f)
- throws IOException {
- return new RemoteIterator<LocatedFileStatus>() {
- private Iterator<LocatedFileStatus> iterator = listLocatedFileStatuses(f).iterator();
-
- @Override
- public boolean hasNext() throws IOException {
- return iterator.hasNext();
- }
-
- @Override
- public LocatedFileStatus next() throws IOException {
- return iterator.next();
- }
- };
- }
-
- private List<LocatedFileStatus> listLocatedFileStatuses(Path path) throws IOException {
- statistics.incrementReadOps(1);
- checkAccess();
- path = path.makeQualified(this);
- List<LocatedFileStatus> result = new ArrayList<>();
- String pathname = path.toString();
- String pathnameAsDir = pathname + "/";
- Set<String> dirs = new TreeSet<String>();
- MockFile file = findFile(path);
- if (file != null) {
- result.add(createLocatedStatus(file));
- return result;
- }
- findMatchingLocatedFiles(files, pathnameAsDir, dirs, result);
- findMatchingLocatedFiles(globalFiles, pathnameAsDir, dirs, result);
- // for each directory add it once
- for(String dir: dirs) {
- result.add(createLocatedDirectory(new MockPath(this, pathnameAsDir + dir)));
- }
- return result;
- }
-
- @Override
- public FileStatus[] listStatus(Path path) throws IOException {
- statistics.incrementReadOps(1);
- checkAccess();
- path = path.makeQualified(this);
- List<FileStatus> result = new ArrayList<FileStatus>();
- String pathname = path.toString();
- String pathnameAsDir = pathname + "/";
- Set<String> dirs = new TreeSet<String>();
- MockFile file = findFile(path);
- if (file != null) {
- return new FileStatus[]{createStatus(file)};
- }
- findMatchingFiles(files, pathnameAsDir, dirs, result);
- findMatchingFiles(globalFiles, pathnameAsDir, dirs, result);
- // for each directory add it once
- for(String dir: dirs) {
- result.add(createDirectory(new MockPath(this, pathnameAsDir + dir)));
- }
- return result.toArray(new FileStatus[result.size()]);
- }
-
- private void findMatchingFiles(
- List<MockFile> files, String pathnameAsDir, Set<String> dirs, List<FileStatus> result) {
- for (MockFile file: files) {
- String filename = file.path.toString();
- if (filename.startsWith(pathnameAsDir)) {
- String tail = filename.substring(pathnameAsDir.length());
- int nextSlash = tail.indexOf('/');
- if (nextSlash > 0) {
- dirs.add(tail.substring(0, nextSlash));
- } else {
- result.add(createStatus(file));
- }
- }
- }
- }
-
- private void findMatchingLocatedFiles(
- List<MockFile> files, String pathnameAsDir, Set<String> dirs, List<LocatedFileStatus> result)
- throws IOException {
- for (MockFile file: files) {
- String filename = file.path.toString();
- if (filename.startsWith(pathnameAsDir)) {
- String tail = filename.substring(pathnameAsDir.length());
- int nextSlash = tail.indexOf('/');
- if (nextSlash > 0) {
- dirs.add(tail.substring(0, nextSlash));
- } else {
- result.add(createLocatedStatus(file));
- }
- }
- }
- }
-
- @Override
- public void setWorkingDirectory(Path path) {
- workingDir = path;
- }
-
- @Override
- public Path getWorkingDirectory() {
- return workingDir;
- }
-
- @Override
- public boolean mkdirs(Path path, FsPermission fsPermission) {
- statistics.incrementWriteOps(1);
- return false;
- }
-
- private FileStatus createStatus(MockFile file) {
- if (fileStatusMap.containsKey(file)) {
- return fileStatusMap.get(file);
- }
- FileStatus fileStatus = new FileStatus(file.length, false, 1, file.blockSize, 0, 0,
- FsPermission.createImmutable((short) 644), "owen", "group",
- file.path);
- fileStatusMap.put(file, fileStatus);
- return fileStatus;
- }
-
- private FileStatus createDirectory(Path dir) {
- return new FileStatus(0, true, 0, 0, 0, 0,
- FsPermission.createImmutable((short) 755), "owen", "group", dir);
- }
-
- private LocatedFileStatus createLocatedStatus(MockFile file) throws IOException {
- FileStatus fileStatus = createStatus(file);
- return new LocatedFileStatus(fileStatus,
- getFileBlockLocationsImpl(fileStatus, 0, fileStatus.getLen(), false));
- }
-
- private LocatedFileStatus createLocatedDirectory(Path dir) throws IOException {
- FileStatus fileStatus = createDirectory(dir);
- return new LocatedFileStatus(fileStatus,
- getFileBlockLocationsImpl(fileStatus, 0, fileStatus.getLen(), false));
- }
-
- @Override
- public FileStatus getFileStatus(Path path) throws IOException {
- statistics.incrementReadOps(1);
- checkAccess();
- path = path.makeQualified(this);
- String pathnameAsDir = path.toString() + "/";
- MockFile file = findFile(path);
- if (file != null) return createStatus(file);
- for (MockFile dir : files) {
- if (dir.path.toString().startsWith(pathnameAsDir)) {
- return createDirectory(path);
- }
- }
- for (MockFile dir : globalFiles) {
- if (dir.path.toString().startsWith(pathnameAsDir)) {
- return createDirectory(path);
- }
- }
- throw new FileNotFoundException("File " + path + " does not exist");
- }
-
- @Override
- public BlockLocation[] getFileBlockLocations(FileStatus stat,
- long start, long len) throws IOException {
- return getFileBlockLocationsImpl(stat, start, len, true);
- }
-
- private BlockLocation[] getFileBlockLocationsImpl(final FileStatus stat, final long start,
- final long len,
- final boolean updateStats) throws IOException {
- if (updateStats) {
- statistics.incrementReadOps(1);
- }
- checkAccess();
- List<BlockLocation> result = new ArrayList<BlockLocation>();
- MockFile file = findFile(stat.getPath());
- if (file != null) {
- for(MockBlock block: file.blocks) {
- if (OrcInputFormat.SplitGenerator.getOverlap(block.offset,
- block.length, start, len) > 0) {
- String[] topology = new String[block.hosts.length];
- for(int i=0; i < topology.length; ++i) {
- topology[i] = "/rack/ " + block.hosts[i];
- }
- result.add(new BlockLocation(block.hosts, block.hosts,
- topology, block.offset, block.length));
- }
- }
- return result.toArray(new BlockLocation[result.size()]);
- }
- return new BlockLocation[0];
- }
-
- @Override
- public String toString() {
- StringBuilder buffer = new StringBuilder();
- buffer.append("mockFs{files:[");
- for(int i=0; i < files.size(); ++i) {
- if (i != 0) {
- buffer.append(", ");
- }
- buffer.append(files.get(i));
- }
- buffer.append("]}");
- return buffer.toString();
- }
-
- public static void addGlobalFile(MockFile mockFile) {
- globalFiles.add(mockFile);
- }
-
- public static void clearGlobalFiles() {
- globalFiles.clear();
- }
- }
-
static void fill(DataOutputBuffer out, long length) throws IOException {
for(int i=0; i < length; ++i) {
out.write(0);
@@ -2569,6 +2031,7 @@ public class TestInputOutputFormat {
conf.setClass("fs.mock.impl", MockFileSystem.class, FileSystem.class);
String badUser = UserGroupInformation.getCurrentUser().getShortUserName() + "-foo";
MockFileSystem.setBlockedUgi(badUser);
+ // TODO: could we instead get FS from path here and add normal files for every UGI?
MockFileSystem.clearGlobalFiles();
OrcInputFormat.Context.resetThreadPool(); // We need the size above to take effect.
try {