Merge 34687b5 into e5a9726

waggle-dance-api/src/main/java/com/hotels/bdp/waggledance/api/model/AbstractMetaStore.java

@@ -46,6 +46,7 @@ public abstract class AbstractMetaStore {
   private String databasePrefix;
   private List<String> writableDatabaseWhitelist;
   private List<String> mappedDatabases;
+  private @Valid List<MappedTables> mappedTables;
   private Map<String, String> databaseNameMapping = Collections.emptyMap();
   private @NotBlank String name;
   private @NotBlank String remoteMetaStoreUris;
@@ -165,6 +166,14 @@ public void setMappedDatabases(List<String> mappedDatabases) {
     this.mappedDatabases = mappedDatabases;
   }
 
+  public List<MappedTables> getMappedTables() {
+    return mappedTables;
+  }
+
+  public void setMappedTables(List<MappedTables> mappedTables) {
+    this.mappedTables = mappedTables;
+  }
+
   public Map<String, String> getDatabaseNameMapping() {
     return databaseNameMapping;
   }

waggle-dance-api/src/main/java/com/hotels/bdp/waggledance/api/model/MappedTables.java

@@ -0,0 +1,50 @@
+/**
+ * Copyright (C) 2016-2020 Expedia, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.hotels.bdp.waggledance.api.model;
+
+import java.util.List;
+
+import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.NotEmpty;
+
+public class MappedTables {
+  @NotBlank private String database;
+  @NotEmpty private List<String> mappedTables;
+
+  public MappedTables() {
+  }
+
+  public MappedTables(String database, List<String> mappedTables) {
+    this.database = database;
+    this.mappedTables = mappedTables;
+  }
+
+  public String getDatabase() {
+    return database;
+  }
+
+  public void setDatabase(String database) {
+    this.database = database;
+  }
+
+  public List<String> getMappedTables() {
+    return mappedTables;
+  }
+
+  public void setMappedTables(List<String> mappedTables) {
+    this.mappedTables = mappedTables;
+  }
+}

waggle-dance-api/src/test/java/com/hotels/bdp/waggledance/api/model/AbstractMetaStoreTest.java

@@ -35,6 +35,7 @@
 import org.springframework.validation.beanvalidation.LocalValidatorFactoryBean;
 
 import com.google.common.collect.HashBiMap;
+import com.google.common.collect.Lists;
 
 import com.hotels.hcommon.hive.metastore.client.tunnelling.MetastoreTunnel;
 
@@ -173,6 +174,72 @@ public void emptyMappedDatabases() {
     assertThat(metaStore.getMappedDatabases().size(), is(0));
   }
 
+  @Test
+  public void mappedTables() {
+    MappedTables mappedTables1 = new MappedTables("db1", Lists.newArrayList("tbl1"));
+    MappedTables mappedTables2 = new MappedTables("db2", Lists.newArrayList("tbl2"));
+    List<MappedTables> mappedTables = Lists.newArrayList(mappedTables1, mappedTables2);
+    metaStore.setMappedTables(mappedTables);
+    assertThat(metaStore.getMappedTables(), is(mappedTables));
+
+    Set<ConstraintViolation<T>> violations = validator.validate(metaStore);
+    assertThat(violations.size(), is(0));
+  }
+
+  @Test
+  public void mappedTablesEmptyDbInvalid() {
+    MappedTables mappedTables = new MappedTables("", Lists.newArrayList("tbl1"));
+    metaStore.setMappedTables(Lists.newArrayList(mappedTables));
+
+    Set<ConstraintViolation<T>> violations = validator.validate(metaStore);
+    assertThat(violations.size(), is(1));
+  }
+
+  @Test
+  public void mappedTablesNullDbInvalid() {
+    MappedTables mappedTables = new MappedTables(null, Lists.newArrayList("tbl1"));
+    metaStore.setMappedTables(Lists.newArrayList(mappedTables));
+
+    Set<ConstraintViolation<T>> violations = validator.validate(metaStore);
+    assertThat(violations.size(), is(1));
+  }
+
+  @Test
+  public void mappedTablesNullTblInvalid() {
+    MappedTables mappedTables = new MappedTables("valid_db", null);
+    metaStore.setMappedTables(Lists.newArrayList(mappedTables));
+
+    Set<ConstraintViolation<T>> violations = validator.validate(metaStore);
+    assertThat(violations.size(), is(1));
+  }
+
+  @Test
+  public void mappedTablesEmptyTblsInvalid() {
+    MappedTables mappedTables = new MappedTables("valid_db", Lists.newArrayList() );
+    metaStore.setMappedTables(Lists.newArrayList(mappedTables));
+
+    Set<ConstraintViolation<T>> violations = validator.validate(metaStore);
+    assertThat(violations.size(), is(1));
+  }
+
+  @Test
+  public void nullMappedTables() {
+    metaStore.setMappedTables(null);
+    assertThat(metaStore.getMappedTables(), is(nullValue()));
+
+    Set<ConstraintViolation<T>> violations = validator.validate(metaStore);
+    assertThat(violations.size(), is(0));
+  }
+
+  @Test
+  public void emptyMappedTables() {
+    metaStore.setMappedTables(Collections.emptyList());
+    assertThat(metaStore.getMappedTables().size(), is(0));
+
+    Set<ConstraintViolation<T>> violations = validator.validate(metaStore);
+    assertThat(violations.size(), is(0));
+  }
+
   @Test
   public void setDatabasesNameMapping() throws Exception {
     Map<String, String> mapping = new HashMap<>();

waggle-dance-api/src/test/java/com/hotels/bdp/waggledance/api/model/FederatedMetaStoreTest.java

@@ -72,7 +72,7 @@ public void nullDatabasePrefix() {
 
   @Test
   public void toJson() throws Exception {
-    String expected = "{\"accessControlType\":\"READ_ONLY\",\"connectionType\":\"DIRECT\",\"databaseNameMapping\":{},\"databasePrefix\":\"name_\",\"federationType\":\"FEDERATED\",\"latency\":0,\"mappedDatabases\":null,\"metastoreTunnel\":null,\"name\":\"name\",\"remoteMetaStoreUris\":\"uri\",\"status\":\"UNKNOWN\",\"writableDatabaseWhiteList\":[]}";
+    String expected = "{\"accessControlType\":\"READ_ONLY\",\"connectionType\":\"DIRECT\",\"databaseNameMapping\":{},\"databasePrefix\":\"name_\",\"federationType\":\"FEDERATED\",\"latency\":0,\"mappedDatabases\":null,\"mappedTables\":null,\"metastoreTunnel\":null,\"name\":\"name\",\"remoteMetaStoreUris\":\"uri\",\"status\":\"UNKNOWN\",\"writableDatabaseWhiteList\":[]}";
     ObjectMapper mapper = new ObjectMapper();
     // Sorting to get deterministic test behaviour
     mapper.enable(MapperFeature.SORT_PROPERTIES_ALPHABETICALLY);

waggle-dance-api/src/test/java/com/hotels/bdp/waggledance/api/model/PrimaryMetaStoreTest.java

@@ -89,7 +89,7 @@ public void nonEmptyDatabasePrefix() {
 
   @Test
   public void toJson() throws Exception {
-    String expected = "{\"accessControlType\":\"READ_ONLY\",\"connectionType\":\"DIRECT\",\"databaseNameMapping\":{},\"databasePrefix\":\"\",\"federationType\":\"PRIMARY\",\"latency\":0,\"mappedDatabases\":null,\"metastoreTunnel\":null,\"name\":\"name\",\"remoteMetaStoreUris\":\"uri\",\"status\":\"UNKNOWN\",\"writableDatabaseWhiteList\":[]}";
+    String expected = "{\"accessControlType\":\"READ_ONLY\",\"connectionType\":\"DIRECT\",\"databaseNameMapping\":{},\"databasePrefix\":\"\",\"federationType\":\"PRIMARY\",\"latency\":0,\"mappedDatabases\":null,\"mappedTables\":null,\"metastoreTunnel\":null,\"name\":\"name\",\"remoteMetaStoreUris\":\"uri\",\"status\":\"UNKNOWN\",\"writableDatabaseWhiteList\":[]}";
     ObjectMapper mapper = new ObjectMapper();
     // Sorting to get deterministic test behaviour
     mapper.enable(MapperFeature.SORT_PROPERTIES_ALPHABETICALLY);

waggle-dance-core/src/main/java/com/hotels/bdp/waggledance/mapping/model/MetaStoreMappingFactoryImpl.java

@@ -60,7 +60,7 @@ private CloseableThriftHiveMetastoreIface createClient(AbstractMetaStore metaSto
       return metaStoreClientFactory.newInstance(metaStore);
     } catch (Exception e) {
       LOG.error("Can't create a client for metastore '{}':", metaStore.getName(), e);
-      return newUnreachableMetatstoreClient(metaStore);
+      return newUnreachableMetastoreClient(metaStore);
     }
   }
 
@@ -85,7 +85,7 @@ public String prefixNameFor(AbstractMetaStore federatedMetaStore) {
     return prefixNamingStrategy.apply(federatedMetaStore);
   }
 
-  private CloseableThriftHiveMetastoreIface newUnreachableMetatstoreClient(AbstractMetaStore metaStore) {
+  private CloseableThriftHiveMetastoreIface newUnreachableMetastoreClient(AbstractMetaStore metaStore) {
     return (CloseableThriftHiveMetastoreIface) Proxy
         .newProxyInstance(getClass().getClassLoader(), new Class[] { CloseableThriftHiveMetastoreIface.class },
             new UnreachableMetastoreClientInvocationHandler(metaStore.getName()));

waggle-dance-core/src/main/java/com/hotels/bdp/waggledance/mapping/service/DatabaseMappingService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2016-2019 Expedia, Inc.
+ * Copyright (C) 2016-2020 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -37,6 +37,11 @@ public interface DatabaseMappingService extends Closeable {
    */
   DatabaseMapping databaseMapping(@NotNull String databaseName) throws NoSuchObjectException;
 
+  void checkTable(String databaseName, String tableName,
+      DatabaseMapping mapping) throws NoSuchObjectException;
+
+  List<String> filterTables(String databaseName, List<String> tableNames, DatabaseMapping mapping);
+
   PanopticOperationHandler getPanopticOperationHandler();
 
   List<DatabaseMapping> getDatabaseMappings();

waggle-dance-core/src/main/java/com/hotels/bdp/waggledance/mapping/service/impl/MonitoredDatabaseMappingService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2016-2019 Expedia, Inc.
+ * Copyright (C) 2016-2020 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -50,6 +50,17 @@ public DatabaseMapping databaseMapping(@NotNull String databaseName) throws NoSu
     return databaseMapping;
   }
 
+  @Override
+  public void checkTable(String databaseName, String tableName,
+      DatabaseMapping mapping) throws NoSuchObjectException {
+      wrapped.checkTable(databaseName, tableName, mapping);
+    }
+
+  @Override
+  public List<String> filterTables(String databaseName, List<String> tableNames, DatabaseMapping mapping) {
+    return wrapped.filterTables(databaseName, tableNames, mapping);
+  }
+
   @Override
   public PanopticOperationHandler getPanopticOperationHandler() {
     PanopticOperationHandler handler = wrapped.getPanopticOperationHandler();

waggle-dance-core/src/main/java/com/hotels/bdp/waggledance/mapping/service/impl/PrefixBasedDatabaseMappingService.java

@@ -44,6 +44,7 @@
 import com.hotels.bdp.waggledance.api.WaggleDanceException;
 import com.hotels.bdp.waggledance.api.model.AbstractMetaStore;
 import com.hotels.bdp.waggledance.api.model.FederationType;
+import com.hotels.bdp.waggledance.api.model.MappedTables;
 import com.hotels.bdp.waggledance.mapping.model.DatabaseMapping;
 import com.hotels.bdp.waggledance.mapping.model.DatabaseMappingImpl;
 import com.hotels.bdp.waggledance.mapping.model.MetaStoreMapping;
@@ -67,6 +68,8 @@ public class PrefixBasedDatabaseMappingService implements MappingEventListener {
   private final QueryMapping queryMapping;
   private final Map<String, DatabaseMapping> mappingsByPrefix;
   private final Map<String, Whitelist> mappedDbByPrefix;
+  private final Map<String, Map<String, Whitelist>> mappedTblByPrefix;
+
   private DatabaseMapping primaryDatabaseMapping;
 
   public PrefixBasedDatabaseMappingService(
@@ -77,6 +80,7 @@ public PrefixBasedDatabaseMappingService(
     this.queryMapping = queryMapping;
     mappingsByPrefix = Collections.synchronizedMap(new LinkedHashMap<>());
     mappedDbByPrefix = new ConcurrentHashMap<>();
+    mappedTblByPrefix = new ConcurrentHashMap<>();
     for (AbstractMetaStore abstractMetaStore : initialMetastores) {
       add(abstractMetaStore);
     }
@@ -93,6 +97,16 @@ private void add(AbstractMetaStore metaStore) {
     mappingsByPrefix.put(metaStoreMapping.getDatabasePrefix(), databaseMapping);
     Whitelist mappedDbWhitelist = new Whitelist(metaStore.getMappedDatabases());
     mappedDbByPrefix.put(metaStoreMapping.getDatabasePrefix(), mappedDbWhitelist);
+
+    List<MappedTables> mappedTables = metaStore.getMappedTables();
+    if (mappedTables != null) {
+    Map<String, Whitelist> mappedTblByDb = new ConcurrentHashMap<>();
+    for (MappedTables mapping : mappedTables) {
+      Whitelist tableWhiteList = new Whitelist(mapping.getMappedTables());
+      mappedTblByDb.put(mapping.getDatabase(), tableWhiteList);
+    }
+    mappedTblByPrefix.put(metaStoreMapping.getDatabasePrefix(), mappedTblByDb);
+    }
   }
 
   private DatabaseMapping createDatabaseMapping(MetaStoreMapping metaStoreMapping) {
@@ -161,7 +175,7 @@ private boolean includeInResults(MetaStoreMapping metaStoreMapping) {
 
   private boolean includeInResults(MetaStoreMapping metaStoreMapping, String prefixedDatabaseName) {
     return includeInResults(metaStoreMapping)
-        && isWhitelisted(metaStoreMapping.getDatabasePrefix(),
+        && isDbWhitelisted(metaStoreMapping.getDatabasePrefix(),
             metaStoreMapping.transformInboundDatabaseName(prefixedDatabaseName));
   }
 
@@ -202,6 +216,42 @@ public DatabaseMapping databaseMapping(@NotNull String databaseName) throws NoSu
         "Waggle Dance error no database mapping available tried to map database '" + databaseName + "'");
   }
 
+  @Override
+  public void checkTable(String databaseName, String tableName, DatabaseMapping mapping)
+      throws NoSuchObjectException {
+    String databasePrefix = mapping.getDatabasePrefix();
+    String transformedDbName = mapping.transformInboundDatabaseName(databaseName);
+    if (!isTableWhitelisted(databasePrefix, transformedDbName, tableName)) {
+      throw new NoSuchObjectException(String.format("%s.%s table not found", databaseName, tableName));
+    }
+  }
+
+  @Override
+  public List<String> filterTables(String databaseName, List<String> tableNames, DatabaseMapping mapping) {
+    List<String> filteredTables = new ArrayList<>();
+    String databasePrefix = mapping.getDatabasePrefix();
+    String transformedDb = mapping.transformInboundDatabaseName(databaseName);
+    for (String table: tableNames)
+    if (isTableWhitelisted(databasePrefix, transformedDb, table)) {
+      filteredTables.add(table);
+    }
+    return filteredTables;
+  }
+
+  private boolean isTableWhitelisted(String databasePrefix, String database, String table) {
+    Map<String, Whitelist> dbToTblWhitelist = mappedTblByPrefix.get(databasePrefix);
+    if (dbToTblWhitelist == null) {
+      // Accept everything
+      return true;
+    }
+    Whitelist tblWhitelist = dbToTblWhitelist.get(database);
+    if (tblWhitelist == null) {
+      // Accept everything
+      return true;
+    }
+    return tblWhitelist.contains(table);
+  }
+
   @Override
   public List<DatabaseMapping> getDatabaseMappings() {
     Builder<DatabaseMapping> builder = ImmutableList.builder();
@@ -234,14 +284,14 @@ private Map<DatabaseMapping, String> databaseMappingsByDbPattern(@NotNull String
   private List<String> getMappedWhitelistedDatabases(List<String> databases, DatabaseMapping mapping) {
     List<String> mappedDatabases = new ArrayList<>();
     for (String database : databases) {
-      if (isWhitelisted(mapping.getDatabasePrefix(), database)) {
+      if (isDbWhitelisted(mapping.getDatabasePrefix(), database)) {
         mappedDatabases.addAll(mapping.transformOutboundDatabaseNameMultiple(database));
       }
     }
     return mappedDatabases;
   }
 
-  private boolean isWhitelisted(String databasePrefix, String database) {
+  private boolean isDbWhitelisted(String databasePrefix, String database) {
     Whitelist whitelist = mappedDbByPrefix.get(databasePrefix);
     if (whitelist == null) {
       // Accept everything
@@ -258,17 +308,21 @@ public PanopticOperationHandler getPanopticOperationHandler() {
       public List<TableMeta> getTableMeta(String db_patterns, String tbl_patterns, List<String> tbl_types) {
         Map<DatabaseMapping, String> databaseMappingsForPattern = databaseMappingsByDbPattern(db_patterns);
 
-        BiFunction<TableMeta, DatabaseMapping, Boolean> filter = (tableMeta, mapping) -> isWhitelisted(
-            mapping.getDatabasePrefix(), tableMeta.getDbName());
-
+        BiFunction<TableMeta, DatabaseMapping, Boolean> filter = (tableMeta, mapping) -> {
+          String dbPrefix = mapping.getDatabasePrefix();
+          String dbName = tableMeta.getDbName();
+          boolean databaseAllowed = isDbWhitelisted(dbPrefix, dbName);
+          boolean tableAllowed = isTableWhitelisted(dbPrefix, dbName, tableMeta.getTableName());
+          return databaseAllowed && tableAllowed;
+        };
         return super.getTableMeta(tbl_patterns, tbl_types, databaseMappingsForPattern, filter);
       }
 
       @Override
       public List<String> getAllDatabases(String databasePattern) {
         Map<DatabaseMapping, String> databaseMappingsForPattern = databaseMappingsByDbPattern(databasePattern);
 
-        BiFunction<String, DatabaseMapping, Boolean> filter = (database, mapping) -> isWhitelisted(
+        BiFunction<String, DatabaseMapping, Boolean> filter = (database, mapping) -> isDbWhitelisted(
             mapping.getDatabasePrefix(), database);
 
         return super.getAllDatabases(databaseMappingsForPattern, filter);