@Test public void testListFilesInFolderRecursive() throws Exception { String folder = "list-files-rec"; String[] nestedFolders = new String[]{"list-files-child-1", "list-files-child-2"}; String[] originalFiles = new String[]{"a-list-3.txt", "b-list-3.txt", "c-list-3.txt"}; List<String> expectedResultList = new ArrayList<>(); for (String childFolder : nestedFolders) { String folderName = folder + DELIMITER + childFolder; for (String fileName : originalFiles) { createEmptyFile(folderName, fileName); expectedResultList.add(String.format(FILE_FORMAT, SCHEME, BUCKET, folderName + DELIMITER + fileName)); } } String[] actualFiles = _s3PinotFS.listFiles(URI.create(String.format(FILE_FORMAT, SCHEME, BUCKET, folder)), true); actualFiles = Arrays.stream(actualFiles).filter(x -> x.contains("list-3")).toArray(String[]::new); Assert.assertEquals(actualFiles.length, expectedResultList.size()); Assert.assertTrue(Arrays.equals(expectedResultList.toArray(), actualFiles)); }
@Override public String[] listFiles(URI fileUri, boolean recursive) throws IOException { try { ImmutableList.Builder<String> builder = ImmutableList.builder(); String continuationToken = null; boolean isDone = false; String prefix = normalizeToDirectoryPrefix(fileUri); while(!isDone) { ListObjectsV2Request.Builder listObjectsV2RequestBuilder = ListObjectsV2Request.builder().bucket(fileUri.getHost()); if (!prefix.equals(DELIMITER)) { listObjectsV2RequestBuilder = listObjectsV2RequestBuilder.prefix(prefix); } if (!recursive) { listObjectsV2RequestBuilder = listObjectsV2RequestBuilder.delimiter(DELIMITER); } if (continuationToken != null) { listObjectsV2RequestBuilder.continuationToken(continuationToken); } ListObjectsV2Request listObjectsV2Request = listObjectsV2RequestBuilder.build(); LOGGER.debug("Trying to send ListObjectsV2Request {}", listObjectsV2Request); ListObjectsV2Response listObjectsV2Response = _s3Client.listObjectsV2(listObjectsV2Request); LOGGER.debug("Getting ListObjectsV2Response: {}", listObjectsV2Response); List<S3Object> filesReturned = listObjectsV2Response.contents(); filesReturned.stream().forEach(object -> { if (!object.key().equals(fileUri.getPath()) && !object.key().endsWith(DELIMITER)) { String fileKey = object.key(); if (fileKey.startsWith(DELIMITER)) { fileKey = fileKey.substring(1); } builder.add(S3_SCHEME + fileUri.getHost() + DELIMITER + fileKey); } }); isDone = !listObjectsV2Response.isTruncated(); continuationToken = listObjectsV2Response.nextContinuationToken(); } String[] listedFiles = builder.build().toArray(new String[0]); LOGGER.info("Listed {} files from URI: {}, is recursive: {}", listedFiles.length, fileUri, recursive); return listedFiles; } catch (Throwable t) { throw new IOException(t); } }
S3PinotFS extends PinotFS { @Override public String[] listFiles(URI fileUri, boolean recursive) throws IOException { try { ImmutableList.Builder<String> builder = ImmutableList.builder(); String continuationToken = null; boolean isDone = false; String prefix = normalizeToDirectoryPrefix(fileUri); while(!isDone) { ListObjectsV2Request.Builder listObjectsV2RequestBuilder = ListObjectsV2Request.builder().bucket(fileUri.getHost()); if (!prefix.equals(DELIMITER)) { listObjectsV2RequestBuilder = listObjectsV2RequestBuilder.prefix(prefix); } if (!recursive) { listObjectsV2RequestBuilder = listObjectsV2RequestBuilder.delimiter(DELIMITER); } if (continuationToken != null) { listObjectsV2RequestBuilder.continuationToken(continuationToken); } ListObjectsV2Request listObjectsV2Request = listObjectsV2RequestBuilder.build(); LOGGER.debug("Trying to send ListObjectsV2Request {}", listObjectsV2Request); ListObjectsV2Response listObjectsV2Response = _s3Client.listObjectsV2(listObjectsV2Request); LOGGER.debug("Getting ListObjectsV2Response: {}", listObjectsV2Response); List<S3Object> filesReturned = listObjectsV2Response.contents(); filesReturned.stream().forEach(object -> { if (!object.key().equals(fileUri.getPath()) && !object.key().endsWith(DELIMITER)) { String fileKey = object.key(); if (fileKey.startsWith(DELIMITER)) { fileKey = fileKey.substring(1); } builder.add(S3_SCHEME + fileUri.getHost() + DELIMITER + fileKey); } }); isDone = !listObjectsV2Response.isTruncated(); continuationToken = listObjectsV2Response.nextContinuationToken(); } String[] listedFiles = builder.build().toArray(new String[0]); LOGGER.info("Listed {} files from URI: {}, is recursive: {}", listedFiles.length, fileUri, recursive); return listedFiles; } catch (Throwable t) { throw new IOException(t); } } }
S3PinotFS extends PinotFS { @Override public String[] listFiles(URI fileUri, boolean recursive) throws IOException { try { ImmutableList.Builder<String> builder = ImmutableList.builder(); String continuationToken = null; boolean isDone = false; String prefix = normalizeToDirectoryPrefix(fileUri); while(!isDone) { ListObjectsV2Request.Builder listObjectsV2RequestBuilder = ListObjectsV2Request.builder().bucket(fileUri.getHost()); if (!prefix.equals(DELIMITER)) { listObjectsV2RequestBuilder = listObjectsV2RequestBuilder.prefix(prefix); } if (!recursive) { listObjectsV2RequestBuilder = listObjectsV2RequestBuilder.delimiter(DELIMITER); } if (continuationToken != null) { listObjectsV2RequestBuilder.continuationToken(continuationToken); } ListObjectsV2Request listObjectsV2Request = listObjectsV2RequestBuilder.build(); LOGGER.debug("Trying to send ListObjectsV2Request {}", listObjectsV2Request); ListObjectsV2Response listObjectsV2Response = _s3Client.listObjectsV2(listObjectsV2Request); LOGGER.debug("Getting ListObjectsV2Response: {}", listObjectsV2Response); List<S3Object> filesReturned = listObjectsV2Response.contents(); filesReturned.stream().forEach(object -> { if (!object.key().equals(fileUri.getPath()) && !object.key().endsWith(DELIMITER)) { String fileKey = object.key(); if (fileKey.startsWith(DELIMITER)) { fileKey = fileKey.substring(1); } builder.add(S3_SCHEME + fileUri.getHost() + DELIMITER + fileKey); } }); isDone = !listObjectsV2Response.isTruncated(); continuationToken = listObjectsV2Response.nextContinuationToken(); } String[] listedFiles = builder.build().toArray(new String[0]); LOGGER.info("Listed {} files from URI: {}, is recursive: {}", listedFiles.length, fileUri, recursive); return listedFiles; } catch (Throwable t) { throw new IOException(t); } }  }
S3PinotFS extends PinotFS { @Override public String[] listFiles(URI fileUri, boolean recursive) throws IOException { try { ImmutableList.Builder<String> builder = ImmutableList.builder(); String continuationToken = null; boolean isDone = false; String prefix = normalizeToDirectoryPrefix(fileUri); while(!isDone) { ListObjectsV2Request.Builder listObjectsV2RequestBuilder = ListObjectsV2Request.builder().bucket(fileUri.getHost()); if (!prefix.equals(DELIMITER)) { listObjectsV2RequestBuilder = listObjectsV2RequestBuilder.prefix(prefix); } if (!recursive) { listObjectsV2RequestBuilder = listObjectsV2RequestBuilder.delimiter(DELIMITER); } if (continuationToken != null) { listObjectsV2RequestBuilder.continuationToken(continuationToken); } ListObjectsV2Request listObjectsV2Request = listObjectsV2RequestBuilder.build(); LOGGER.debug("Trying to send ListObjectsV2Request {}", listObjectsV2Request); ListObjectsV2Response listObjectsV2Response = _s3Client.listObjectsV2(listObjectsV2Request); LOGGER.debug("Getting ListObjectsV2Response: {}", listObjectsV2Response); List<S3Object> filesReturned = listObjectsV2Response.contents(); filesReturned.stream().forEach(object -> { if (!object.key().equals(fileUri.getPath()) && !object.key().endsWith(DELIMITER)) { String fileKey = object.key(); if (fileKey.startsWith(DELIMITER)) { fileKey = fileKey.substring(1); } builder.add(S3_SCHEME + fileUri.getHost() + DELIMITER + fileKey); } }); isDone = !listObjectsV2Response.isTruncated(); continuationToken = listObjectsV2Response.nextContinuationToken(); } String[] listedFiles = builder.build().toArray(new String[0]); LOGGER.info("Listed {} files from URI: {}, is recursive: {}", listedFiles.length, fileUri, recursive); return listedFiles; } catch (Throwable t) { throw new IOException(t); } }  @Override void init(PinotConfiguration config); void init(S3Client s3Client); @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); @Override void close(); }
S3PinotFS extends PinotFS { @Override public String[] listFiles(URI fileUri, boolean recursive) throws IOException { try { ImmutableList.Builder<String> builder = ImmutableList.builder(); String continuationToken = null; boolean isDone = false; String prefix = normalizeToDirectoryPrefix(fileUri); while(!isDone) { ListObjectsV2Request.Builder listObjectsV2RequestBuilder = ListObjectsV2Request.builder().bucket(fileUri.getHost()); if (!prefix.equals(DELIMITER)) { listObjectsV2RequestBuilder = listObjectsV2RequestBuilder.prefix(prefix); } if (!recursive) { listObjectsV2RequestBuilder = listObjectsV2RequestBuilder.delimiter(DELIMITER); } if (continuationToken != null) { listObjectsV2RequestBuilder.continuationToken(continuationToken); } ListObjectsV2Request listObjectsV2Request = listObjectsV2RequestBuilder.build(); LOGGER.debug("Trying to send ListObjectsV2Request {}", listObjectsV2Request); ListObjectsV2Response listObjectsV2Response = _s3Client.listObjectsV2(listObjectsV2Request); LOGGER.debug("Getting ListObjectsV2Response: {}", listObjectsV2Response); List<S3Object> filesReturned = listObjectsV2Response.contents(); filesReturned.stream().forEach(object -> { if (!object.key().equals(fileUri.getPath()) && !object.key().endsWith(DELIMITER)) { String fileKey = object.key(); if (fileKey.startsWith(DELIMITER)) { fileKey = fileKey.substring(1); } builder.add(S3_SCHEME + fileUri.getHost() + DELIMITER + fileKey); } }); isDone = !listObjectsV2Response.isTruncated(); continuationToken = listObjectsV2Response.nextContinuationToken(); } String[] listedFiles = builder.build().toArray(new String[0]); LOGGER.info("Listed {} files from URI: {}, is recursive: {}", listedFiles.length, fileUri, recursive); return listedFiles; } catch (Throwable t) { throw new IOException(t); } }  @Override void init(PinotConfiguration config); void init(S3Client s3Client); @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); @Override void close(); static final String ACCESS_KEY; static final String SECRET_KEY; static final String REGION; static final String ENDPOINT; static final String S3_SCHEME; }