@Test public void delete_sharetemp() throws Exception { HadoopDataSourceProfile shareTempProfile = new HadoopDataSourceProfile( conf, profile.getId(), profile.getContextPath(), profile.getFileSystemPath(), new Path(profile.getFileSystemPath(), "_TEMP")); HadoopDataSourceCore core = new HadoopDataSourceCore(shareTempProfile); File onProd = new File(mapping, "file.txt"); File onTemp = new File(mapping, "_TEMP/temp.txt"); put(onProd, "production"); put(onTemp, "temporary"); assertThat(onProd.exists(), is(true)); assertThat(onTemp.exists(), is(true)); boolean result = core.delete("", FilePattern.compile("**/*"), true, counter); assertThat(result, is(true)); assertThat(onProd.exists(), is(false)); assertThat(onTemp.exists(), is(true)); }
@Override public boolean delete( String basePath, ResourcePattern resourcePattern, boolean recursive, Counter counter) throws IOException, InterruptedException { assert basePath.startsWith("/") == false; if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start deleting files (id={0}, path={1}, resource={2}, recursive={3})", profile.getId(), basePath, resourcePattern, recursive)); } FilePattern pattern = validate(resourcePattern); HadoopDataSourceProfile p = profile; FileSystem fs = p.getFileSystem(); Path root = p.getFileSystemPath(); Path base = append(root, basePath); List<FileStatus> stats = HadoopDataSourceUtil.search(fs, base, pattern); Path temporary = p.getTemporaryFileSystemPath(); stats = normalize(stats, root, temporary); if (recursive) { stats = HadoopDataSourceUtil.onlyMinimalCovered(stats); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Process deleting files (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } boolean succeed = true; for (FileStatus stat : stats) { if (LOG.isTraceEnabled()) { LOG.trace(MessageFormat.format( "Deleting file (id={0}, path={1}, recursive={2})", profile.getId(), stat.getPath(), recursive)); } if (recursive == false && stat.isDirectory()) { LOG.info(MessageFormat.format( "Skip deleting directory (id={0}, path={1})", profile.getId(), stat.getPath())); } else { counter.add(1); succeed &= fs.delete(stat.getPath(), recursive); } } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish deleting files (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } return succeed; }
HadoopDataSourceCore implements DirectDataSource { @Override public boolean delete( String basePath, ResourcePattern resourcePattern, boolean recursive, Counter counter) throws IOException, InterruptedException { assert basePath.startsWith("/") == false; if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start deleting files (id={0}, path={1}, resource={2}, recursive={3})", profile.getId(), basePath, resourcePattern, recursive)); } FilePattern pattern = validate(resourcePattern); HadoopDataSourceProfile p = profile; FileSystem fs = p.getFileSystem(); Path root = p.getFileSystemPath(); Path base = append(root, basePath); List<FileStatus> stats = HadoopDataSourceUtil.search(fs, base, pattern); Path temporary = p.getTemporaryFileSystemPath(); stats = normalize(stats, root, temporary); if (recursive) { stats = HadoopDataSourceUtil.onlyMinimalCovered(stats); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Process deleting files (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } boolean succeed = true; for (FileStatus stat : stats) { if (LOG.isTraceEnabled()) { LOG.trace(MessageFormat.format( "Deleting file (id={0}, path={1}, recursive={2})", profile.getId(), stat.getPath(), recursive)); } if (recursive == false && stat.isDirectory()) { LOG.info(MessageFormat.format( "Skip deleting directory (id={0}, path={1})", profile.getId(), stat.getPath())); } else { counter.add(1); succeed &= fs.delete(stat.getPath(), recursive); } } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish deleting files (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } return succeed; } }
HadoopDataSourceCore implements DirectDataSource { @Override public boolean delete( String basePath, ResourcePattern resourcePattern, boolean recursive, Counter counter) throws IOException, InterruptedException { assert basePath.startsWith("/") == false; if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start deleting files (id={0}, path={1}, resource={2}, recursive={3})", profile.getId(), basePath, resourcePattern, recursive)); } FilePattern pattern = validate(resourcePattern); HadoopDataSourceProfile p = profile; FileSystem fs = p.getFileSystem(); Path root = p.getFileSystemPath(); Path base = append(root, basePath); List<FileStatus> stats = HadoopDataSourceUtil.search(fs, base, pattern); Path temporary = p.getTemporaryFileSystemPath(); stats = normalize(stats, root, temporary); if (recursive) { stats = HadoopDataSourceUtil.onlyMinimalCovered(stats); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Process deleting files (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } boolean succeed = true; for (FileStatus stat : stats) { if (LOG.isTraceEnabled()) { LOG.trace(MessageFormat.format( "Deleting file (id={0}, path={1}, recursive={2})", profile.getId(), stat.getPath(), recursive)); } if (recursive == false && stat.isDirectory()) { LOG.info(MessageFormat.format( "Skip deleting directory (id={0}, path={1})", profile.getId(), stat.getPath())); } else { counter.add(1); succeed &= fs.delete(stat.getPath(), recursive); } } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish deleting files (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } return succeed; } HadoopDataSourceCore(HadoopDataSourceProfile profile); }
HadoopDataSourceCore implements DirectDataSource { @Override public boolean delete( String basePath, ResourcePattern resourcePattern, boolean recursive, Counter counter) throws IOException, InterruptedException { assert basePath.startsWith("/") == false; if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start deleting files (id={0}, path={1}, resource={2}, recursive={3})", profile.getId(), basePath, resourcePattern, recursive)); } FilePattern pattern = validate(resourcePattern); HadoopDataSourceProfile p = profile; FileSystem fs = p.getFileSystem(); Path root = p.getFileSystemPath(); Path base = append(root, basePath); List<FileStatus> stats = HadoopDataSourceUtil.search(fs, base, pattern); Path temporary = p.getTemporaryFileSystemPath(); stats = normalize(stats, root, temporary); if (recursive) { stats = HadoopDataSourceUtil.onlyMinimalCovered(stats); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Process deleting files (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } boolean succeed = true; for (FileStatus stat : stats) { if (LOG.isTraceEnabled()) { LOG.trace(MessageFormat.format( "Deleting file (id={0}, path={1}, recursive={2})", profile.getId(), stat.getPath(), recursive)); } if (recursive == false && stat.isDirectory()) { LOG.info(MessageFormat.format( "Skip deleting directory (id={0}, path={1})", profile.getId(), stat.getPath())); } else { counter.add(1); succeed &= fs.delete(stat.getPath(), recursive); } } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish deleting files (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } return succeed; } HadoopDataSourceCore(HadoopDataSourceProfile profile); @Override String path(String basePath, ResourcePattern resourcePattern); @Override String path(String basePath); @Override List<DirectInputFragment> findInputFragments(
            DataDefinition<T> definition,
            String basePath,
            ResourcePattern resourcePattern); @Override ModelInput<T> openInput(
            DataDefinition<T> definition,
            DirectInputFragment fragment,
            Counter counter); @Override ModelOutput<T> openOutput(
            OutputAttemptContext context,
            DataDefinition<T> definition,
            String basePath,
            String resourcePath,
            Counter counter); @Override List<ResourceInfo> list(
            String basePath,
            ResourcePattern resourcePattern,
            Counter counter); @Override boolean delete(
            String basePath,
            ResourcePattern resourcePattern,
            boolean recursive,
            Counter counter); @Override void setupAttemptOutput(OutputAttemptContext context); @Override void commitAttemptOutput(OutputAttemptContext context); @Override void cleanupAttemptOutput(OutputAttemptContext context); @Override void setupTransactionOutput(OutputTransactionContext context); @Override void commitTransactionOutput(OutputTransactionContext context); @Override void cleanupTransactionOutput(OutputTransactionContext context); @Override Optional<T> findProperty(Class<T> propertyType); }
HadoopDataSourceCore implements DirectDataSource { @Override public boolean delete( String basePath, ResourcePattern resourcePattern, boolean recursive, Counter counter) throws IOException, InterruptedException { assert basePath.startsWith("/") == false; if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start deleting files (id={0}, path={1}, resource={2}, recursive={3})", profile.getId(), basePath, resourcePattern, recursive)); } FilePattern pattern = validate(resourcePattern); HadoopDataSourceProfile p = profile; FileSystem fs = p.getFileSystem(); Path root = p.getFileSystemPath(); Path base = append(root, basePath); List<FileStatus> stats = HadoopDataSourceUtil.search(fs, base, pattern); Path temporary = p.getTemporaryFileSystemPath(); stats = normalize(stats, root, temporary); if (recursive) { stats = HadoopDataSourceUtil.onlyMinimalCovered(stats); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Process deleting files (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } boolean succeed = true; for (FileStatus stat : stats) { if (LOG.isTraceEnabled()) { LOG.trace(MessageFormat.format( "Deleting file (id={0}, path={1}, recursive={2})", profile.getId(), stat.getPath(), recursive)); } if (recursive == false && stat.isDirectory()) { LOG.info(MessageFormat.format( "Skip deleting directory (id={0}, path={1})", profile.getId(), stat.getPath())); } else { counter.add(1); succeed &= fs.delete(stat.getPath(), recursive); } } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish deleting files (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } return succeed; } HadoopDataSourceCore(HadoopDataSourceProfile profile); @Override String path(String basePath, ResourcePattern resourcePattern); @Override String path(String basePath); @Override List<DirectInputFragment> findInputFragments(
            DataDefinition<T> definition,
            String basePath,
            ResourcePattern resourcePattern); @Override ModelInput<T> openInput(
            DataDefinition<T> definition,
            DirectInputFragment fragment,
            Counter counter); @Override ModelOutput<T> openOutput(
            OutputAttemptContext context,
            DataDefinition<T> definition,
            String basePath,
            String resourcePath,
            Counter counter); @Override List<ResourceInfo> list(
            String basePath,
            ResourcePattern resourcePattern,
            Counter counter); @Override boolean delete(
            String basePath,
            ResourcePattern resourcePattern,
            boolean recursive,
            Counter counter); @Override void setupAttemptOutput(OutputAttemptContext context); @Override void commitAttemptOutput(OutputAttemptContext context); @Override void cleanupAttemptOutput(OutputAttemptContext context); @Override void setupTransactionOutput(OutputTransactionContext context); @Override void commitTransactionOutput(OutputTransactionContext context); @Override void cleanupTransactionOutput(OutputTransactionContext context); @Override Optional<T> findProperty(Class<T> propertyType);  }