@Test public void testTouchFilesInFolder() throws Exception { String folder = "my-files"; String[] originalFiles = new String[]{"a-touch.txt", "b-touch.txt", "c-touch.txt"}; for (String fileName : originalFiles) { String fileNameWithFolder = folder + DELIMITER + fileName; _s3PinotFS.touch(URI.create(String.format(FILE_FORMAT, SCHEME, BUCKET, fileNameWithFolder))); } ListObjectsV2Response listObjectsV2Response = _s3Client.listObjectsV2(S3TestUtils.getListObjectRequest(BUCKET, folder, false)); String[] response = listObjectsV2Response.contents().stream().map(S3Object::key).filter(x -> x.contains("touch")) .toArray(String[]::new); Assert.assertEquals(response.length, originalFiles.length); Assert.assertTrue(Arrays.equals(response, Arrays.stream(originalFiles).map(x -> folder + DELIMITER + x).toArray())); }
@Override public boolean touch(URI uri) throws IOException { try { HeadObjectResponse s3ObjectMetadata = getS3ObjectMetadata(uri); String encodedUrl = null; try { encodedUrl = URLEncoder.encode(uri.getHost() + uri.getPath(), StandardCharsets.UTF_8.toString()); } catch (UnsupportedEncodingException e) { throw new RuntimeException(e); } String path = sanitizePath(uri.getPath()); Map<String, String> mp = new HashMap<>(); mp.put("lastModified", String.valueOf(System.currentTimeMillis())); CopyObjectRequest request = CopyObjectRequest.builder().copySource(encodedUrl).destinationBucket(uri.getHost()).destinationKey(path) .metadata(mp).metadataDirective(MetadataDirective.REPLACE).build(); _s3Client.copyObject(request); long newUpdateTime = getS3ObjectMetadata(uri).lastModified().toEpochMilli(); return newUpdateTime > s3ObjectMetadata.lastModified().toEpochMilli(); } catch (NoSuchKeyException e) { String path = sanitizePath(uri.getPath()); _s3Client.putObject(PutObjectRequest.builder().bucket(uri.getHost()).key(path).build(), RequestBody.fromBytes(new byte[0])); return true; } catch (S3Exception e) { throw new IOException(e); } }
S3PinotFS extends PinotFS { @Override public boolean touch(URI uri) throws IOException { try { HeadObjectResponse s3ObjectMetadata = getS3ObjectMetadata(uri); String encodedUrl = null; try { encodedUrl = URLEncoder.encode(uri.getHost() + uri.getPath(), StandardCharsets.UTF_8.toString()); } catch (UnsupportedEncodingException e) { throw new RuntimeException(e); } String path = sanitizePath(uri.getPath()); Map<String, String> mp = new HashMap<>(); mp.put("lastModified", String.valueOf(System.currentTimeMillis())); CopyObjectRequest request = CopyObjectRequest.builder().copySource(encodedUrl).destinationBucket(uri.getHost()).destinationKey(path) .metadata(mp).metadataDirective(MetadataDirective.REPLACE).build(); _s3Client.copyObject(request); long newUpdateTime = getS3ObjectMetadata(uri).lastModified().toEpochMilli(); return newUpdateTime > s3ObjectMetadata.lastModified().toEpochMilli(); } catch (NoSuchKeyException e) { String path = sanitizePath(uri.getPath()); _s3Client.putObject(PutObjectRequest.builder().bucket(uri.getHost()).key(path).build(), RequestBody.fromBytes(new byte[0])); return true; } catch (S3Exception e) { throw new IOException(e); } } }
S3PinotFS extends PinotFS { @Override public boolean touch(URI uri) throws IOException { try { HeadObjectResponse s3ObjectMetadata = getS3ObjectMetadata(uri); String encodedUrl = null; try { encodedUrl = URLEncoder.encode(uri.getHost() + uri.getPath(), StandardCharsets.UTF_8.toString()); } catch (UnsupportedEncodingException e) { throw new RuntimeException(e); } String path = sanitizePath(uri.getPath()); Map<String, String> mp = new HashMap<>(); mp.put("lastModified", String.valueOf(System.currentTimeMillis())); CopyObjectRequest request = CopyObjectRequest.builder().copySource(encodedUrl).destinationBucket(uri.getHost()).destinationKey(path) .metadata(mp).metadataDirective(MetadataDirective.REPLACE).build(); _s3Client.copyObject(request); long newUpdateTime = getS3ObjectMetadata(uri).lastModified().toEpochMilli(); return newUpdateTime > s3ObjectMetadata.lastModified().toEpochMilli(); } catch (NoSuchKeyException e) { String path = sanitizePath(uri.getPath()); _s3Client.putObject(PutObjectRequest.builder().bucket(uri.getHost()).key(path).build(), RequestBody.fromBytes(new byte[0])); return true; } catch (S3Exception e) { throw new IOException(e); } }  }
S3PinotFS extends PinotFS { @Override public boolean touch(URI uri) throws IOException { try { HeadObjectResponse s3ObjectMetadata = getS3ObjectMetadata(uri); String encodedUrl = null; try { encodedUrl = URLEncoder.encode(uri.getHost() + uri.getPath(), StandardCharsets.UTF_8.toString()); } catch (UnsupportedEncodingException e) { throw new RuntimeException(e); } String path = sanitizePath(uri.getPath()); Map<String, String> mp = new HashMap<>(); mp.put("lastModified", String.valueOf(System.currentTimeMillis())); CopyObjectRequest request = CopyObjectRequest.builder().copySource(encodedUrl).destinationBucket(uri.getHost()).destinationKey(path) .metadata(mp).metadataDirective(MetadataDirective.REPLACE).build(); _s3Client.copyObject(request); long newUpdateTime = getS3ObjectMetadata(uri).lastModified().toEpochMilli(); return newUpdateTime > s3ObjectMetadata.lastModified().toEpochMilli(); } catch (NoSuchKeyException e) { String path = sanitizePath(uri.getPath()); _s3Client.putObject(PutObjectRequest.builder().bucket(uri.getHost()).key(path).build(), RequestBody.fromBytes(new byte[0])); return true; } catch (S3Exception e) { throw new IOException(e); } }  @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 boolean touch(URI uri) throws IOException { try { HeadObjectResponse s3ObjectMetadata = getS3ObjectMetadata(uri); String encodedUrl = null; try { encodedUrl = URLEncoder.encode(uri.getHost() + uri.getPath(), StandardCharsets.UTF_8.toString()); } catch (UnsupportedEncodingException e) { throw new RuntimeException(e); } String path = sanitizePath(uri.getPath()); Map<String, String> mp = new HashMap<>(); mp.put("lastModified", String.valueOf(System.currentTimeMillis())); CopyObjectRequest request = CopyObjectRequest.builder().copySource(encodedUrl).destinationBucket(uri.getHost()).destinationKey(path) .metadata(mp).metadataDirective(MetadataDirective.REPLACE).build(); _s3Client.copyObject(request); long newUpdateTime = getS3ObjectMetadata(uri).lastModified().toEpochMilli(); return newUpdateTime > s3ObjectMetadata.lastModified().toEpochMilli(); } catch (NoSuchKeyException e) { String path = sanitizePath(uri.getPath()); _s3Client.putObject(PutObjectRequest.builder().bucket(uri.getHost()).key(path).build(), RequestBody.fromBytes(new byte[0])); return true; } catch (S3Exception e) { throw new IOException(e); } }  @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; }