@Test public void testHoodiePaths() throws Exception { final String p1 = "2017/01/01"; final String p2 = "2017/01/02"; testTable.addCommit("001") .withBaseFilesInPartition(p1, "f1", "f2") .withBaseFilesInPartition(p2, "f3") .addCommit("002") .withBaseFilesInPartition(p1, "f2") .addInflightCommit("003") .withBaseFilesInPartition(p2, "f3") .addRequestedCompaction("004"); assertTrue(pathFilter.accept(testTable.forCommit("002").getBaseFilePath(p1, "f2"))); assertFalse(pathFilter.accept(testTable.forCommit("003").getBaseFilePath(p2, "f3"))); assertFalse(pathFilter.accept(testTable.forCommit("003").getBaseFilePath(p1, "f3"))); assertFalse(pathFilter.accept(testTable.getCommitFilePath("001"))); assertFalse(pathFilter.accept(testTable.getCommitFilePath("002"))); assertFalse(pathFilter.accept(testTable.getInflightCommitFilePath("003"))); assertFalse(pathFilter.accept(testTable.getRequestedCompactionFilePath("004"))); assertFalse(pathFilter.accept(new Path("file: assertFalse(pathFilter.accept(new Path("file: assertEquals(1, pathFilter.metaClientCache.size()); }
@Override public boolean accept(Path path) { if (LOG.isDebugEnabled()) { LOG.debug("Checking acceptance for path " + path); } Path folder = null; try { if (fs == null) { fs = path.getFileSystem(conf.get()); } folder = path.getParent(); if (nonHoodiePathCache.contains(folder.toString())) { if (LOG.isDebugEnabled()) { LOG.debug("Accepting non-hoodie path from cache: " + path); } return true; } if (hoodiePathCache.containsKey(folder.toString())) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("%s Hoodie path checked against cache, accept => %s \n", path, hoodiePathCache.get(folder.toString()).contains(path))); } return hoodiePathCache.get(folder.toString()).contains(path); } String filePath = path.toString(); if (filePath.contains("/" + HoodieTableMetaClient.METAFOLDER_NAME + "/") || filePath.endsWith("/" + HoodieTableMetaClient.METAFOLDER_NAME)) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("Skipping Hoodie Metadata file %s \n", filePath)); } return false; } Path baseDir; if (HoodiePartitionMetadata.hasPartitionMetadata(fs, folder)) { HoodiePartitionMetadata metadata = new HoodiePartitionMetadata(fs, folder); metadata.readFromFS(); baseDir = HoodieHiveUtils.getNthParent(folder, metadata.getPartitionDepth()); } else { baseDir = safeGetParentsParent(folder); } if (baseDir != null) { try { HoodieTableMetaClient metaClient = metaClientCache.get(baseDir.toString()); if (null == metaClient) { metaClient = new HoodieTableMetaClient(fs.getConf(), baseDir.toString(), true); metaClientCache.put(baseDir.toString(), metaClient); } HoodieTableFileSystemView fsView = new HoodieTableFileSystemView(metaClient, metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants(), fs.listStatus(folder)); List<HoodieBaseFile> latestFiles = fsView.getLatestBaseFiles().collect(Collectors.toList()); if (!hoodiePathCache.containsKey(folder.toString())) { hoodiePathCache.put(folder.toString(), new HashSet<>()); } LOG.info("Based on hoodie metadata from base path: " + baseDir.toString() + ", caching " + latestFiles.size() + " files under " + folder); for (HoodieBaseFile lfile : latestFiles) { hoodiePathCache.get(folder.toString()).add(new Path(lfile.getPath())); } if (LOG.isDebugEnabled()) { LOG.debug(String.format("%s checked after cache population, accept => %s \n", path, hoodiePathCache.get(folder.toString()).contains(path))); } return hoodiePathCache.get(folder.toString()).contains(path); } catch (TableNotFoundException e) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("(1) Caching non-hoodie path under %s \n", folder.toString())); } nonHoodiePathCache.add(folder.toString()); return true; } } else { if (LOG.isDebugEnabled()) { LOG.debug(String.format("(2) Caching non-hoodie path under %s \n", folder.toString())); } nonHoodiePathCache.add(folder.toString()); return true; } } catch (Exception e) { String msg = "Error checking path :" + path + ", under folder: " + folder; LOG.error(msg, e); throw new HoodieException(msg, e); } }
HoodieROTablePathFilter implements Configurable, PathFilter, Serializable { @Override public boolean accept(Path path) { if (LOG.isDebugEnabled()) { LOG.debug("Checking acceptance for path " + path); } Path folder = null; try { if (fs == null) { fs = path.getFileSystem(conf.get()); } folder = path.getParent(); if (nonHoodiePathCache.contains(folder.toString())) { if (LOG.isDebugEnabled()) { LOG.debug("Accepting non-hoodie path from cache: " + path); } return true; } if (hoodiePathCache.containsKey(folder.toString())) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("%s Hoodie path checked against cache, accept => %s \n", path, hoodiePathCache.get(folder.toString()).contains(path))); } return hoodiePathCache.get(folder.toString()).contains(path); } String filePath = path.toString(); if (filePath.contains("/" + HoodieTableMetaClient.METAFOLDER_NAME + "/") || filePath.endsWith("/" + HoodieTableMetaClient.METAFOLDER_NAME)) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("Skipping Hoodie Metadata file %s \n", filePath)); } return false; } Path baseDir; if (HoodiePartitionMetadata.hasPartitionMetadata(fs, folder)) { HoodiePartitionMetadata metadata = new HoodiePartitionMetadata(fs, folder); metadata.readFromFS(); baseDir = HoodieHiveUtils.getNthParent(folder, metadata.getPartitionDepth()); } else { baseDir = safeGetParentsParent(folder); } if (baseDir != null) { try { HoodieTableMetaClient metaClient = metaClientCache.get(baseDir.toString()); if (null == metaClient) { metaClient = new HoodieTableMetaClient(fs.getConf(), baseDir.toString(), true); metaClientCache.put(baseDir.toString(), metaClient); } HoodieTableFileSystemView fsView = new HoodieTableFileSystemView(metaClient, metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants(), fs.listStatus(folder)); List<HoodieBaseFile> latestFiles = fsView.getLatestBaseFiles().collect(Collectors.toList()); if (!hoodiePathCache.containsKey(folder.toString())) { hoodiePathCache.put(folder.toString(), new HashSet<>()); } LOG.info("Based on hoodie metadata from base path: " + baseDir.toString() + ", caching " + latestFiles.size() + " files under " + folder); for (HoodieBaseFile lfile : latestFiles) { hoodiePathCache.get(folder.toString()).add(new Path(lfile.getPath())); } if (LOG.isDebugEnabled()) { LOG.debug(String.format("%s checked after cache population, accept => %s \n", path, hoodiePathCache.get(folder.toString()).contains(path))); } return hoodiePathCache.get(folder.toString()).contains(path); } catch (TableNotFoundException e) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("(1) Caching non-hoodie path under %s \n", folder.toString())); } nonHoodiePathCache.add(folder.toString()); return true; } } else { if (LOG.isDebugEnabled()) { LOG.debug(String.format("(2) Caching non-hoodie path under %s \n", folder.toString())); } nonHoodiePathCache.add(folder.toString()); return true; } } catch (Exception e) { String msg = "Error checking path :" + path + ", under folder: " + folder; LOG.error(msg, e); throw new HoodieException(msg, e); } } }
HoodieROTablePathFilter implements Configurable, PathFilter, Serializable { @Override public boolean accept(Path path) { if (LOG.isDebugEnabled()) { LOG.debug("Checking acceptance for path " + path); } Path folder = null; try { if (fs == null) { fs = path.getFileSystem(conf.get()); } folder = path.getParent(); if (nonHoodiePathCache.contains(folder.toString())) { if (LOG.isDebugEnabled()) { LOG.debug("Accepting non-hoodie path from cache: " + path); } return true; } if (hoodiePathCache.containsKey(folder.toString())) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("%s Hoodie path checked against cache, accept => %s \n", path, hoodiePathCache.get(folder.toString()).contains(path))); } return hoodiePathCache.get(folder.toString()).contains(path); } String filePath = path.toString(); if (filePath.contains("/" + HoodieTableMetaClient.METAFOLDER_NAME + "/") || filePath.endsWith("/" + HoodieTableMetaClient.METAFOLDER_NAME)) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("Skipping Hoodie Metadata file %s \n", filePath)); } return false; } Path baseDir; if (HoodiePartitionMetadata.hasPartitionMetadata(fs, folder)) { HoodiePartitionMetadata metadata = new HoodiePartitionMetadata(fs, folder); metadata.readFromFS(); baseDir = HoodieHiveUtils.getNthParent(folder, metadata.getPartitionDepth()); } else { baseDir = safeGetParentsParent(folder); } if (baseDir != null) { try { HoodieTableMetaClient metaClient = metaClientCache.get(baseDir.toString()); if (null == metaClient) { metaClient = new HoodieTableMetaClient(fs.getConf(), baseDir.toString(), true); metaClientCache.put(baseDir.toString(), metaClient); } HoodieTableFileSystemView fsView = new HoodieTableFileSystemView(metaClient, metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants(), fs.listStatus(folder)); List<HoodieBaseFile> latestFiles = fsView.getLatestBaseFiles().collect(Collectors.toList()); if (!hoodiePathCache.containsKey(folder.toString())) { hoodiePathCache.put(folder.toString(), new HashSet<>()); } LOG.info("Based on hoodie metadata from base path: " + baseDir.toString() + ", caching " + latestFiles.size() + " files under " + folder); for (HoodieBaseFile lfile : latestFiles) { hoodiePathCache.get(folder.toString()).add(new Path(lfile.getPath())); } if (LOG.isDebugEnabled()) { LOG.debug(String.format("%s checked after cache population, accept => %s \n", path, hoodiePathCache.get(folder.toString()).contains(path))); } return hoodiePathCache.get(folder.toString()).contains(path); } catch (TableNotFoundException e) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("(1) Caching non-hoodie path under %s \n", folder.toString())); } nonHoodiePathCache.add(folder.toString()); return true; } } else { if (LOG.isDebugEnabled()) { LOG.debug(String.format("(2) Caching non-hoodie path under %s \n", folder.toString())); } nonHoodiePathCache.add(folder.toString()); return true; } } catch (Exception e) { String msg = "Error checking path :" + path + ", under folder: " + folder; LOG.error(msg, e); throw new HoodieException(msg, e); } } HoodieROTablePathFilter();  HoodieROTablePathFilter(Configuration conf); }
HoodieROTablePathFilter implements Configurable, PathFilter, Serializable { @Override public boolean accept(Path path) { if (LOG.isDebugEnabled()) { LOG.debug("Checking acceptance for path " + path); } Path folder = null; try { if (fs == null) { fs = path.getFileSystem(conf.get()); } folder = path.getParent(); if (nonHoodiePathCache.contains(folder.toString())) { if (LOG.isDebugEnabled()) { LOG.debug("Accepting non-hoodie path from cache: " + path); } return true; } if (hoodiePathCache.containsKey(folder.toString())) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("%s Hoodie path checked against cache, accept => %s \n", path, hoodiePathCache.get(folder.toString()).contains(path))); } return hoodiePathCache.get(folder.toString()).contains(path); } String filePath = path.toString(); if (filePath.contains("/" + HoodieTableMetaClient.METAFOLDER_NAME + "/") || filePath.endsWith("/" + HoodieTableMetaClient.METAFOLDER_NAME)) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("Skipping Hoodie Metadata file %s \n", filePath)); } return false; } Path baseDir; if (HoodiePartitionMetadata.hasPartitionMetadata(fs, folder)) { HoodiePartitionMetadata metadata = new HoodiePartitionMetadata(fs, folder); metadata.readFromFS(); baseDir = HoodieHiveUtils.getNthParent(folder, metadata.getPartitionDepth()); } else { baseDir = safeGetParentsParent(folder); } if (baseDir != null) { try { HoodieTableMetaClient metaClient = metaClientCache.get(baseDir.toString()); if (null == metaClient) { metaClient = new HoodieTableMetaClient(fs.getConf(), baseDir.toString(), true); metaClientCache.put(baseDir.toString(), metaClient); } HoodieTableFileSystemView fsView = new HoodieTableFileSystemView(metaClient, metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants(), fs.listStatus(folder)); List<HoodieBaseFile> latestFiles = fsView.getLatestBaseFiles().collect(Collectors.toList()); if (!hoodiePathCache.containsKey(folder.toString())) { hoodiePathCache.put(folder.toString(), new HashSet<>()); } LOG.info("Based on hoodie metadata from base path: " + baseDir.toString() + ", caching " + latestFiles.size() + " files under " + folder); for (HoodieBaseFile lfile : latestFiles) { hoodiePathCache.get(folder.toString()).add(new Path(lfile.getPath())); } if (LOG.isDebugEnabled()) { LOG.debug(String.format("%s checked after cache population, accept => %s \n", path, hoodiePathCache.get(folder.toString()).contains(path))); } return hoodiePathCache.get(folder.toString()).contains(path); } catch (TableNotFoundException e) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("(1) Caching non-hoodie path under %s \n", folder.toString())); } nonHoodiePathCache.add(folder.toString()); return true; } } else { if (LOG.isDebugEnabled()) { LOG.debug(String.format("(2) Caching non-hoodie path under %s \n", folder.toString())); } nonHoodiePathCache.add(folder.toString()); return true; } } catch (Exception e) { String msg = "Error checking path :" + path + ", under folder: " + folder; LOG.error(msg, e); throw new HoodieException(msg, e); } } HoodieROTablePathFilter();  HoodieROTablePathFilter(Configuration conf); @Override boolean accept(Path path); @Override void setConf(Configuration conf); @Override Configuration getConf(); }
HoodieROTablePathFilter implements Configurable, PathFilter, Serializable { @Override public boolean accept(Path path) { if (LOG.isDebugEnabled()) { LOG.debug("Checking acceptance for path " + path); } Path folder = null; try { if (fs == null) { fs = path.getFileSystem(conf.get()); } folder = path.getParent(); if (nonHoodiePathCache.contains(folder.toString())) { if (LOG.isDebugEnabled()) { LOG.debug("Accepting non-hoodie path from cache: " + path); } return true; } if (hoodiePathCache.containsKey(folder.toString())) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("%s Hoodie path checked against cache, accept => %s \n", path, hoodiePathCache.get(folder.toString()).contains(path))); } return hoodiePathCache.get(folder.toString()).contains(path); } String filePath = path.toString(); if (filePath.contains("/" + HoodieTableMetaClient.METAFOLDER_NAME + "/") || filePath.endsWith("/" + HoodieTableMetaClient.METAFOLDER_NAME)) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("Skipping Hoodie Metadata file %s \n", filePath)); } return false; } Path baseDir; if (HoodiePartitionMetadata.hasPartitionMetadata(fs, folder)) { HoodiePartitionMetadata metadata = new HoodiePartitionMetadata(fs, folder); metadata.readFromFS(); baseDir = HoodieHiveUtils.getNthParent(folder, metadata.getPartitionDepth()); } else { baseDir = safeGetParentsParent(folder); } if (baseDir != null) { try { HoodieTableMetaClient metaClient = metaClientCache.get(baseDir.toString()); if (null == metaClient) { metaClient = new HoodieTableMetaClient(fs.getConf(), baseDir.toString(), true); metaClientCache.put(baseDir.toString(), metaClient); } HoodieTableFileSystemView fsView = new HoodieTableFileSystemView(metaClient, metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants(), fs.listStatus(folder)); List<HoodieBaseFile> latestFiles = fsView.getLatestBaseFiles().collect(Collectors.toList()); if (!hoodiePathCache.containsKey(folder.toString())) { hoodiePathCache.put(folder.toString(), new HashSet<>()); } LOG.info("Based on hoodie metadata from base path: " + baseDir.toString() + ", caching " + latestFiles.size() + " files under " + folder); for (HoodieBaseFile lfile : latestFiles) { hoodiePathCache.get(folder.toString()).add(new Path(lfile.getPath())); } if (LOG.isDebugEnabled()) { LOG.debug(String.format("%s checked after cache population, accept => %s \n", path, hoodiePathCache.get(folder.toString()).contains(path))); } return hoodiePathCache.get(folder.toString()).contains(path); } catch (TableNotFoundException e) { if (LOG.isDebugEnabled()) { LOG.debug(String.format("(1) Caching non-hoodie path under %s \n", folder.toString())); } nonHoodiePathCache.add(folder.toString()); return true; } } else { if (LOG.isDebugEnabled()) { LOG.debug(String.format("(2) Caching non-hoodie path under %s \n", folder.toString())); } nonHoodiePathCache.add(folder.toString()); return true; } } catch (Exception e) { String msg = "Error checking path :" + path + ", under folder: " + folder; LOG.error(msg, e); throw new HoodieException(msg, e); } } HoodieROTablePathFilter();  HoodieROTablePathFilter(Configuration conf); @Override boolean accept(Path path); @Override void setConf(Configuration conf); @Override Configuration getConf();  }