@Test public void testTableFilter() throws Exception { List<HiveDatasetFinder.DbAndTable> dbAndTables = Lists.newArrayList(); dbAndTables.add(new HiveDatasetFinder.DbAndTable("db1", "table1")); dbAndTables.add(new HiveDatasetFinder.DbAndTable("db1", "table2")); dbAndTables.add(new DatePartitionTable("db1", "table3")); dbAndTables.add(new DatePartitionTable("db1", "table4")); HiveMetastoreClientPool pool = getTestPool(dbAndTables); Properties properties = new Properties(); properties.put(HiveDatasetFinder.HIVE_DATASET_PREFIX + "." + WhitelistBlacklist.WHITELIST, ""); properties.put(HiveDatasetFinder.TABLE_FILTER, TableTypeFilter.class.getName()); properties.put(TableTypeFilter.FILTER_TYPE, "snapshot"); HiveDatasetFinder finder = new TestHiveDatasetFinder(FileSystem.getLocal(new Configuration()), properties, pool); List<HiveDataset> datasets = Lists.newArrayList(finder.getDatasetsIterator()); Assert.assertEquals(datasets.size(), 2); Assert.assertTrue(datasets.stream().anyMatch(dataset -> dataset.getDbAndTable().toString().equals("db1.table1"))); Assert.assertTrue(datasets.stream().anyMatch(dataset -> dataset.getDbAndTable().toString().equals("db1.table2"))); properties.put(TableTypeFilter.FILTER_TYPE, "partitioned"); finder = new TestHiveDatasetFinder(FileSystem.getLocal(new Configuration()), properties, pool); datasets = Lists.newArrayList(finder.getDatasetsIterator()); Assert.assertEquals(datasets.size(), 2); Assert.assertTrue(datasets.stream().anyMatch(dataset -> dataset.getDbAndTable().toString().equals("db1.table3"))); Assert.assertTrue(datasets.stream().anyMatch(dataset -> dataset.getDbAndTable().toString().equals("db1.table4"))); }
@Override public Iterator<HiveDataset> getDatasetsIterator() throws IOException { return new AbstractIterator<HiveDataset>() { private Iterator<DbAndTable> tables = getTables().iterator(); @Override protected HiveDataset computeNext() { while (this.tables.hasNext()) { DbAndTable dbAndTable = this.tables.next(); try (AutoReturnableObject<IMetaStoreClient> client = HiveDatasetFinder.this.clientPool.getClient()) { Table table = client.get().getTable(dbAndTable.getDb(), dbAndTable.getTable()); if (tableFilter.isPresent() && !tableFilter.get().apply(table)) { continue; } Config datasetConfig = getDatasetConfig(table); if (ConfigUtils.getBoolean(datasetConfig, HIVE_DATASET_IS_BLACKLISTED_KEY, DEFAULT_HIVE_DATASET_IS_BLACKLISTED_KEY)) { continue; } if (HiveDatasetFinder.this.eventSubmitter.isPresent()) { SlaEventSubmitter.builder().datasetUrn(dbAndTable.toString()) .eventSubmitter(HiveDatasetFinder.this.eventSubmitter.get()).eventName(DATASET_FOUND).build().submit(); } return createHiveDataset(table, datasetConfig); } catch (IllegalArgumentException e) { Throwables.propagate(e); } catch (Throwable t) { log.error(String.format("Failed to create HiveDataset for table %s.%s", dbAndTable.getDb(), dbAndTable.getTable()), t); if (HiveDatasetFinder.this.eventSubmitter.isPresent()) { SlaEventSubmitter.builder().datasetUrn(dbAndTable.toString()) .eventSubmitter(HiveDatasetFinder.this.eventSubmitter.get()).eventName(DATASET_ERROR) .additionalMetadata(FAILURE_CONTEXT, t.toString()).build().submit(); } } } return endOfData(); } }; }
HiveDatasetFinder implements IterableDatasetFinder<HiveDataset> { @Override public Iterator<HiveDataset> getDatasetsIterator() throws IOException { return new AbstractIterator<HiveDataset>() { private Iterator<DbAndTable> tables = getTables().iterator(); @Override protected HiveDataset computeNext() { while (this.tables.hasNext()) { DbAndTable dbAndTable = this.tables.next(); try (AutoReturnableObject<IMetaStoreClient> client = HiveDatasetFinder.this.clientPool.getClient()) { Table table = client.get().getTable(dbAndTable.getDb(), dbAndTable.getTable()); if (tableFilter.isPresent() && !tableFilter.get().apply(table)) { continue; } Config datasetConfig = getDatasetConfig(table); if (ConfigUtils.getBoolean(datasetConfig, HIVE_DATASET_IS_BLACKLISTED_KEY, DEFAULT_HIVE_DATASET_IS_BLACKLISTED_KEY)) { continue; } if (HiveDatasetFinder.this.eventSubmitter.isPresent()) { SlaEventSubmitter.builder().datasetUrn(dbAndTable.toString()) .eventSubmitter(HiveDatasetFinder.this.eventSubmitter.get()).eventName(DATASET_FOUND).build().submit(); } return createHiveDataset(table, datasetConfig); } catch (IllegalArgumentException e) { Throwables.propagate(e); } catch (Throwable t) { log.error(String.format("Failed to create HiveDataset for table %s.%s", dbAndTable.getDb(), dbAndTable.getTable()), t); if (HiveDatasetFinder.this.eventSubmitter.isPresent()) { SlaEventSubmitter.builder().datasetUrn(dbAndTable.toString()) .eventSubmitter(HiveDatasetFinder.this.eventSubmitter.get()).eventName(DATASET_ERROR) .additionalMetadata(FAILURE_CONTEXT, t.toString()).build().submit(); } } } return endOfData(); } }; } }
HiveDatasetFinder implements IterableDatasetFinder<HiveDataset> { @Override public Iterator<HiveDataset> getDatasetsIterator() throws IOException { return new AbstractIterator<HiveDataset>() { private Iterator<DbAndTable> tables = getTables().iterator(); @Override protected HiveDataset computeNext() { while (this.tables.hasNext()) { DbAndTable dbAndTable = this.tables.next(); try (AutoReturnableObject<IMetaStoreClient> client = HiveDatasetFinder.this.clientPool.getClient()) { Table table = client.get().getTable(dbAndTable.getDb(), dbAndTable.getTable()); if (tableFilter.isPresent() && !tableFilter.get().apply(table)) { continue; } Config datasetConfig = getDatasetConfig(table); if (ConfigUtils.getBoolean(datasetConfig, HIVE_DATASET_IS_BLACKLISTED_KEY, DEFAULT_HIVE_DATASET_IS_BLACKLISTED_KEY)) { continue; } if (HiveDatasetFinder.this.eventSubmitter.isPresent()) { SlaEventSubmitter.builder().datasetUrn(dbAndTable.toString()) .eventSubmitter(HiveDatasetFinder.this.eventSubmitter.get()).eventName(DATASET_FOUND).build().submit(); } return createHiveDataset(table, datasetConfig); } catch (IllegalArgumentException e) { Throwables.propagate(e); } catch (Throwable t) { log.error(String.format("Failed to create HiveDataset for table %s.%s", dbAndTable.getDb(), dbAndTable.getTable()), t); if (HiveDatasetFinder.this.eventSubmitter.isPresent()) { SlaEventSubmitter.builder().datasetUrn(dbAndTable.toString()) .eventSubmitter(HiveDatasetFinder.this.eventSubmitter.get()).eventName(DATASET_ERROR) .additionalMetadata(FAILURE_CONTEXT, t.toString()).build().submit(); } } } return endOfData(); } }; } HiveDatasetFinder(FileSystem fs, Properties properties); protected  HiveDatasetFinder(FileSystem fs, Properties properties, ConfigClient configClient);  HiveDatasetFinder(FileSystem fs, Properties properties, EventSubmitter eventSubmitter); protected  HiveDatasetFinder(FileSystem fs, Properties properties, HiveMetastoreClientPool clientPool); protected  HiveDatasetFinder(FileSystem fs, Properties properties, HiveMetastoreClientPool clientPool,
      EventSubmitter eventSubmitter); @SuppressWarnings("unchecked") //SupressWarning justification : CONFIG_STORE_DATASET_URI_BUILDER_CLASS must be of type Function<DbAndTable, String>. //It is safe to throw RuntimeException otherwise protected  HiveDatasetFinder(FileSystem fs, Properties properties, HiveMetastoreClientPool clientPool,
      EventSubmitter eventSubmitter, ConfigClient configClient); }
HiveDatasetFinder implements IterableDatasetFinder<HiveDataset> { @Override public Iterator<HiveDataset> getDatasetsIterator() throws IOException { return new AbstractIterator<HiveDataset>() { private Iterator<DbAndTable> tables = getTables().iterator(); @Override protected HiveDataset computeNext() { while (this.tables.hasNext()) { DbAndTable dbAndTable = this.tables.next(); try (AutoReturnableObject<IMetaStoreClient> client = HiveDatasetFinder.this.clientPool.getClient()) { Table table = client.get().getTable(dbAndTable.getDb(), dbAndTable.getTable()); if (tableFilter.isPresent() && !tableFilter.get().apply(table)) { continue; } Config datasetConfig = getDatasetConfig(table); if (ConfigUtils.getBoolean(datasetConfig, HIVE_DATASET_IS_BLACKLISTED_KEY, DEFAULT_HIVE_DATASET_IS_BLACKLISTED_KEY)) { continue; } if (HiveDatasetFinder.this.eventSubmitter.isPresent()) { SlaEventSubmitter.builder().datasetUrn(dbAndTable.toString()) .eventSubmitter(HiveDatasetFinder.this.eventSubmitter.get()).eventName(DATASET_FOUND).build().submit(); } return createHiveDataset(table, datasetConfig); } catch (IllegalArgumentException e) { Throwables.propagate(e); } catch (Throwable t) { log.error(String.format("Failed to create HiveDataset for table %s.%s", dbAndTable.getDb(), dbAndTable.getTable()), t); if (HiveDatasetFinder.this.eventSubmitter.isPresent()) { SlaEventSubmitter.builder().datasetUrn(dbAndTable.toString()) .eventSubmitter(HiveDatasetFinder.this.eventSubmitter.get()).eventName(DATASET_ERROR) .additionalMetadata(FAILURE_CONTEXT, t.toString()).build().submit(); } } } return endOfData(); } }; } HiveDatasetFinder(FileSystem fs, Properties properties); protected  HiveDatasetFinder(FileSystem fs, Properties properties, ConfigClient configClient);  HiveDatasetFinder(FileSystem fs, Properties properties, EventSubmitter eventSubmitter); protected  HiveDatasetFinder(FileSystem fs, Properties properties, HiveMetastoreClientPool clientPool); protected  HiveDatasetFinder(FileSystem fs, Properties properties, HiveMetastoreClientPool clientPool,
      EventSubmitter eventSubmitter); @SuppressWarnings("unchecked") //SupressWarning justification : CONFIG_STORE_DATASET_URI_BUILDER_CLASS must be of type Function<DbAndTable, String>. //It is safe to throw RuntimeException otherwise protected  HiveDatasetFinder(FileSystem fs, Properties properties, HiveMetastoreClientPool clientPool,
      EventSubmitter eventSubmitter, ConfigClient configClient); Collection<DbAndTable> getTables(); @Override List<HiveDataset> findDatasets(); @Override Iterator<HiveDataset> getDatasetsIterator(); @Override Path commonDatasetRoot(); }
HiveDatasetFinder implements IterableDatasetFinder<HiveDataset> { @Override public Iterator<HiveDataset> getDatasetsIterator() throws IOException { return new AbstractIterator<HiveDataset>() { private Iterator<DbAndTable> tables = getTables().iterator(); @Override protected HiveDataset computeNext() { while (this.tables.hasNext()) { DbAndTable dbAndTable = this.tables.next(); try (AutoReturnableObject<IMetaStoreClient> client = HiveDatasetFinder.this.clientPool.getClient()) { Table table = client.get().getTable(dbAndTable.getDb(), dbAndTable.getTable()); if (tableFilter.isPresent() && !tableFilter.get().apply(table)) { continue; } Config datasetConfig = getDatasetConfig(table); if (ConfigUtils.getBoolean(datasetConfig, HIVE_DATASET_IS_BLACKLISTED_KEY, DEFAULT_HIVE_DATASET_IS_BLACKLISTED_KEY)) { continue; } if (HiveDatasetFinder.this.eventSubmitter.isPresent()) { SlaEventSubmitter.builder().datasetUrn(dbAndTable.toString()) .eventSubmitter(HiveDatasetFinder.this.eventSubmitter.get()).eventName(DATASET_FOUND).build().submit(); } return createHiveDataset(table, datasetConfig); } catch (IllegalArgumentException e) { Throwables.propagate(e); } catch (Throwable t) { log.error(String.format("Failed to create HiveDataset for table %s.%s", dbAndTable.getDb(), dbAndTable.getTable()), t); if (HiveDatasetFinder.this.eventSubmitter.isPresent()) { SlaEventSubmitter.builder().datasetUrn(dbAndTable.toString()) .eventSubmitter(HiveDatasetFinder.this.eventSubmitter.get()).eventName(DATASET_ERROR) .additionalMetadata(FAILURE_CONTEXT, t.toString()).build().submit(); } } } return endOfData(); } }; } HiveDatasetFinder(FileSystem fs, Properties properties); protected  HiveDatasetFinder(FileSystem fs, Properties properties, ConfigClient configClient);  HiveDatasetFinder(FileSystem fs, Properties properties, EventSubmitter eventSubmitter); protected  HiveDatasetFinder(FileSystem fs, Properties properties, HiveMetastoreClientPool clientPool); protected  HiveDatasetFinder(FileSystem fs, Properties properties, HiveMetastoreClientPool clientPool,
      EventSubmitter eventSubmitter); @SuppressWarnings("unchecked") //SupressWarning justification : CONFIG_STORE_DATASET_URI_BUILDER_CLASS must be of type Function<DbAndTable, String>. //It is safe to throw RuntimeException otherwise protected  HiveDatasetFinder(FileSystem fs, Properties properties, HiveMetastoreClientPool clientPool,
      EventSubmitter eventSubmitter, ConfigClient configClient); Collection<DbAndTable> getTables(); @Override List<HiveDataset> findDatasets(); @Override Iterator<HiveDataset> getDatasetsIterator(); @Override Path commonDatasetRoot(); static final String HIVE_DATASET_PREFIX; static final String HIVE_METASTORE_URI_KEY; static final String DB_KEY; static final String TABLE_PATTERN_KEY; static final String DEFAULT_TABLE_PATTERN; static final String TABLE_FILTER; static final String HIVE_DATASET_CONFIG_PREFIX_KEY; static final String HIVE_DATASET_IS_BLACKLISTED_KEY; static final String CONFIG_STORE_DATASET_URI_BUILDER_CLASS; }