@Test public void testCopy() throws IOException { URI baseURI = URI.create(TMP_DIR + "/HadoopPinotFSTest"); try (HadoopPinotFS hadoopFS = new HadoopPinotFS()) { hadoopFS.init(new PinotConfiguration()); hadoopFS.mkdir(new Path(baseURI.getPath(), "src").toUri()); hadoopFS.mkdir(new Path(baseURI.getPath(), "src/dir").toUri()); hadoopFS.touch(new Path(baseURI.getPath(), "src/dir/1").toUri()); hadoopFS.touch(new Path(baseURI.getPath(), "src/dir/2").toUri()); String[] srcFiles = hadoopFS.listFiles(new Path(baseURI.getPath(), "src").toUri(), true); Assert.assertEquals(srcFiles.length, 3); hadoopFS.copy(new Path(baseURI.getPath(), "src").toUri(), new Path(baseURI.getPath(), "dest").toUri()); Assert.assertTrue(hadoopFS.exists(new Path(baseURI.getPath(), "dest").toUri())); Assert.assertTrue(hadoopFS.exists(new Path(baseURI.getPath(), "dest/dir").toUri())); Assert.assertTrue(hadoopFS.exists(new Path(baseURI.getPath(), "dest/dir/1").toUri())); Assert.assertTrue(hadoopFS.exists(new Path(baseURI.getPath(), "dest/dir/2").toUri())); String[] destFiles = hadoopFS.listFiles(new Path(baseURI.getPath(), "dest").toUri(), true); Assert.assertEquals(destFiles.length, 3); hadoopFS.delete(baseURI, true); } }
@Override public boolean copy(URI srcUri, URI dstUri) throws IOException { Path source = new Path(srcUri); Path target = new Path(dstUri); RemoteIterator<FileStatus> sourceFiles = _hadoopFS.listStatusIterator(source); if (sourceFiles != null) { while (sourceFiles.hasNext()) { FileStatus sourceFile = sourceFiles.next(); Path sourceFilePath = sourceFile.getPath(); if (sourceFile.isFile()) { try { FileUtil.copy(_hadoopFS, sourceFilePath, _hadoopFS, new Path(target, sourceFilePath.getName()), false, _hadoopConf); } catch (FileNotFoundException e) { LOGGER.warn("Not found file {}, skipping copying it...", sourceFilePath, e); } } else if (sourceFile.isDirectory()) { try { copy(sourceFilePath.toUri(), new Path(target, sourceFilePath.getName()).toUri()); } catch (FileNotFoundException e) { LOGGER.warn("Not found directory {}, skipping copying it...", sourceFilePath, e); } } } } return true; }
HadoopPinotFS extends PinotFS { @Override public boolean copy(URI srcUri, URI dstUri) throws IOException { Path source = new Path(srcUri); Path target = new Path(dstUri); RemoteIterator<FileStatus> sourceFiles = _hadoopFS.listStatusIterator(source); if (sourceFiles != null) { while (sourceFiles.hasNext()) { FileStatus sourceFile = sourceFiles.next(); Path sourceFilePath = sourceFile.getPath(); if (sourceFile.isFile()) { try { FileUtil.copy(_hadoopFS, sourceFilePath, _hadoopFS, new Path(target, sourceFilePath.getName()), false, _hadoopConf); } catch (FileNotFoundException e) { LOGGER.warn("Not found file {}, skipping copying it...", sourceFilePath, e); } } else if (sourceFile.isDirectory()) { try { copy(sourceFilePath.toUri(), new Path(target, sourceFilePath.getName()).toUri()); } catch (FileNotFoundException e) { LOGGER.warn("Not found directory {}, skipping copying it...", sourceFilePath, e); } } } } return true; } }
HadoopPinotFS extends PinotFS { @Override public boolean copy(URI srcUri, URI dstUri) throws IOException { Path source = new Path(srcUri); Path target = new Path(dstUri); RemoteIterator<FileStatus> sourceFiles = _hadoopFS.listStatusIterator(source); if (sourceFiles != null) { while (sourceFiles.hasNext()) { FileStatus sourceFile = sourceFiles.next(); Path sourceFilePath = sourceFile.getPath(); if (sourceFile.isFile()) { try { FileUtil.copy(_hadoopFS, sourceFilePath, _hadoopFS, new Path(target, sourceFilePath.getName()), false, _hadoopConf); } catch (FileNotFoundException e) { LOGGER.warn("Not found file {}, skipping copying it...", sourceFilePath, e); } } else if (sourceFile.isDirectory()) { try { copy(sourceFilePath.toUri(), new Path(target, sourceFilePath.getName()).toUri()); } catch (FileNotFoundException e) { LOGGER.warn("Not found directory {}, skipping copying it...", sourceFilePath, e); } } } } return true; } HadoopPinotFS(); }
HadoopPinotFS extends PinotFS { @Override public boolean copy(URI srcUri, URI dstUri) throws IOException { Path source = new Path(srcUri); Path target = new Path(dstUri); RemoteIterator<FileStatus> sourceFiles = _hadoopFS.listStatusIterator(source); if (sourceFiles != null) { while (sourceFiles.hasNext()) { FileStatus sourceFile = sourceFiles.next(); Path sourceFilePath = sourceFile.getPath(); if (sourceFile.isFile()) { try { FileUtil.copy(_hadoopFS, sourceFilePath, _hadoopFS, new Path(target, sourceFilePath.getName()), false, _hadoopConf); } catch (FileNotFoundException e) { LOGGER.warn("Not found file {}, skipping copying it...", sourceFilePath, e); } } else if (sourceFile.isDirectory()) { try { copy(sourceFilePath.toUri(), new Path(target, sourceFilePath.getName()).toUri()); } catch (FileNotFoundException e) { LOGGER.warn("Not found directory {}, skipping copying it...", sourceFilePath, e); } } } } return true; } HadoopPinotFS(); @Override void init(PinotConfiguration config); @Override boolean mkdir(URI uri); @Override boolean delete(URI segmentUri, boolean forceDelete); @Override boolean doMove(URI srcUri, URI dstUri); @Override boolean copy(URI srcUri, URI dstUri); @Override boolean exists(URI fileUri); @Override long length(URI fileUri); @Override String[] listFiles(URI fileUri, boolean recursive); @Override void copyToLocalFile(URI srcUri, File dstFile); @Override void copyFromLocalFile(File srcFile, URI dstUri); @Override boolean isDirectory(URI uri); @Override long lastModified(URI uri); @Override boolean touch(URI uri); @Override InputStream open(URI uri); }
HadoopPinotFS extends PinotFS { @Override public boolean copy(URI srcUri, URI dstUri) throws IOException { Path source = new Path(srcUri); Path target = new Path(dstUri); RemoteIterator<FileStatus> sourceFiles = _hadoopFS.listStatusIterator(source); if (sourceFiles != null) { while (sourceFiles.hasNext()) { FileStatus sourceFile = sourceFiles.next(); Path sourceFilePath = sourceFile.getPath(); if (sourceFile.isFile()) { try { FileUtil.copy(_hadoopFS, sourceFilePath, _hadoopFS, new Path(target, sourceFilePath.getName()), false, _hadoopConf); } catch (FileNotFoundException e) { LOGGER.warn("Not found file {}, skipping copying it...", sourceFilePath, e); } } else if (sourceFile.isDirectory()) { try { copy(sourceFilePath.toUri(), new Path(target, sourceFilePath.getName()).toUri()); } catch (FileNotFoundException e) { LOGGER.warn("Not found directory {}, skipping copying it...", sourceFilePath, e); } } } } return true; } HadoopPinotFS(); @Override void init(PinotConfiguration config); @Override boolean mkdir(URI uri); @Override boolean delete(URI segmentUri, boolean forceDelete); @Override boolean doMove(URI srcUri, URI dstUri); @Override boolean copy(URI srcUri, URI dstUri); @Override boolean exists(URI fileUri); @Override long length(URI fileUri); @Override String[] listFiles(URI fileUri, boolean recursive); @Override void copyToLocalFile(URI srcUri, File dstFile); @Override void copyFromLocalFile(File srcFile, URI dstUri); @Override boolean isDirectory(URI uri); @Override long lastModified(URI uri); @Override boolean touch(URI uri); @Override InputStream open(URI uri);  }