Updated tests and modified according to rick's comments

fili-core/src/main/java/com/yahoo/bard/webservice/table/availability/MetricUnionAvailability.java

@@ -80,7 +80,8 @@ public MetricUnionAvailability(@NotNull Set<PhysicalTable> physicalTables, @NotN
                 .map(MetricColumn::getName)
                 .collect(Collectors.toSet());
 
-        // map each metric to its corresponding availability object
+        // Construct a map of availability to its assigned metric
+        // by intersecting its underlying datasource metrics with table configured metrics
         availabilitiesToMetricNames = physicalTables.stream()
                 .map(PhysicalTable::getAvailability)
                 .collect(
@@ -178,8 +179,7 @@ public String toString() {
     private static boolean isMetricUnique(Map<Availability, Set<String>> availabilityToMetricNames) {
         Set<String> uniqueMetrics = new HashSet<>();
 
-        return availabilityToMetricNames.entrySet().stream()
-                .map(Map.Entry::getValue)
+        return availabilityToMetricNames.values().stream()
                 .flatMap(Set::stream)
                 .allMatch(uniqueMetrics::add);
     }

fili-core/src/main/java/com/yahoo/bard/webservice/table/availability/PermissiveAvailability.java

@@ -47,16 +47,13 @@ public PermissiveAvailability(
      */
     @Override
     public SimplifiedIntervalList getAvailableIntervals(PhysicalDataSourceConstraint ignoredConstraint) {
-
         return getAllAvailableIntervals().values().stream()
                 .map(SimplifiedIntervalList::new)
                 .reduce(new SimplifiedIntervalList(), SimplifiedIntervalList::simplifyIntervals);
     }
 
     @Override
     public String toString() {
-        return String.format("PermissiveAvailability with table name = %s",
-                getName().asName()
-        );
+        return String.format("PermissiveAvailability with table name = %s", getName().asName());
     }
 }

fili-core/src/main/java/com/yahoo/bard/webservice/table/resolver/PhysicalDataSourceConstraint.java

@@ -2,12 +2,9 @@
 // Licensed under the terms of the Apache license. Please see LICENSE.md file distributed with this work for terms.
 package com.yahoo.bard.webservice.table.resolver;
 
-import com.yahoo.bard.webservice.data.dimension.Dimension;
 import com.yahoo.bard.webservice.table.PhysicalTableSchema;
-import com.yahoo.bard.webservice.web.ApiFilter;
 
 import java.util.Collections;
-import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -41,7 +38,7 @@ public PhysicalDataSourceConstraint(
     }
 
     /**
-     * Constructor, use with care, beware of caching behavior.
+     * Constructor, use with care, beware of inconsistent caching behavior.
      *
      * @param dataSourceConstraint  Data source constraint containing all the column names as logical names
      * @param allColumnPhysicalNames  The physical names of the columns
@@ -55,43 +52,6 @@ private PhysicalDataSourceConstraint(
 
     }
 
-    /**
-     * Constructor.
-     *
-     * @param requestDimensions  Dimensions contained in request
-     * @param filterDimensions  Filtered dimensions
-     * @param metricDimensions  Metric related dimensions
-     * @param metricNames  Names of metrics
-     * @param allDimensions  Set of all dimension objects
-     * @param allDimensionNames  Set of all dimension names
-     * @param allColumnNames  Set of all column names
-     * @param apiFilters  Map of dimension to its set of API filters
-     * @param allColumnPhysicalNames  Set of all column physical names
-     */
-    protected PhysicalDataSourceConstraint(
-            Set<Dimension> requestDimensions,
-            Set<Dimension> filterDimensions,
-            Set<Dimension> metricDimensions,
-            Set<String> metricNames,
-            Set<Dimension> allDimensions,
-            Set<String> allDimensionNames,
-            Set<String> allColumnNames,
-            Map<Dimension, Set<ApiFilter>> apiFilters,
-            Set<String> allColumnPhysicalNames
-    ) {
-        super(
-                requestDimensions,
-                filterDimensions,
-                metricDimensions,
-                metricNames,
-                allDimensions,
-                allDimensionNames,
-                allColumnNames,
-                apiFilters
-        );
-        this.allColumnPhysicalNames = allColumnPhysicalNames;
-    }
-
     /**
      * Getter for the all column names as physical names.
      *

fili-core/src/test/groovy/com/yahoo/bard/webservice/table/MetricUnionCompositeTableSpec.groovy

@@ -16,7 +16,7 @@ class MetricUnionCompositeTableSpec extends Specification {
 
     def "Constructor throws illegal argument exception on empty physical tables"() {
         when:
-        MetricUnionCompositeTable metricUnionCompositeTable = new MetricUnionCompositeTable(
+        new MetricUnionCompositeTable(
                 tableName,
                 [] as Set,
                 [] as Set,

fili-core/src/test/groovy/com/yahoo/bard/webservice/table/availability/MetricUnionAvailabilitySpec.groovy

@@ -4,9 +4,11 @@ package com.yahoo.bard.webservice.table.availability
 
 import com.yahoo.bard.webservice.data.config.names.TableName
 import com.yahoo.bard.webservice.data.metric.MetricColumn
+import com.yahoo.bard.webservice.table.Column
 import com.yahoo.bard.webservice.table.PhysicalTable
 import com.yahoo.bard.webservice.table.resolver.DataSourceConstraint
 import com.yahoo.bard.webservice.table.resolver.PhysicalDataSourceConstraint
+import com.yahoo.bard.webservice.table.resolver.PhysicalDataSourceConstraintSpec
 import com.yahoo.bard.webservice.util.SimplifiedIntervalList
 
 import org.joda.time.Interval
@@ -24,6 +26,9 @@ class MetricUnionAvailabilitySpec extends Specification {
     String metric1
     String metric2
 
+    Column metricColumn1
+    Column metricColumn2
+
     PhysicalTable physicalTable1
     PhysicalTable physicalTable2
 
@@ -41,6 +46,9 @@ class MetricUnionAvailabilitySpec extends Specification {
         metric1 = 'metric1'
         metric2 = 'metric2'
 
+        metricColumn1 = new MetricColumn(metric1)
+        metricColumn2 = new MetricColumn(metric2)
+
         physicalTable1 = Mock(PhysicalTable)
         physicalTable2 = Mock(PhysicalTable)
 
@@ -62,7 +70,7 @@ class MetricUnionAvailabilitySpec extends Specification {
         MetricColumn tableColumn1 = new MetricColumn("shouldNotBeReturned1")
         MetricColumn tableColumn2 = new MetricColumn("shouldNotBeReturned2")
 
-        metricUnionAvailability = new MetricUnionAvailability(physicalTables, [new MetricColumn(metric1), new MetricColumn(metric2)] as Set)
+        metricUnionAvailability = new MetricUnionAvailability(physicalTables, [metricColumn1, metricColumn2] as Set)
 
         when:
         Map<Availability, Set<String>> availabilitiesToAvailableColumns = metricUnionAvailability.availabilitiesToMetricNames
@@ -84,7 +92,7 @@ class MetricUnionAvailabilitySpec extends Specification {
     }
 
     @Unroll
-    def "checkDuplicateValue returns duplicate metric column names or empty, if no duplicates, from 2 physical tables in the case of #caseDescription"() {
+    def "isMetricUnique returns true if and only if metric is unique across all data sources in the case of #caseDescription"() {
         when:
         boolean actual = MetricUnionAvailability.isMetricUnique(
                 [
@@ -116,7 +124,7 @@ class MetricUnionAvailabilitySpec extends Specification {
         ]
 
         when:
-        metricUnionAvailability = new MetricUnionAvailability(physicalTables, [new MetricColumn(metric1)] as Set)
+        metricUnionAvailability = new MetricUnionAvailability(physicalTables, [metricColumn1] as Set)
 
         then:
         RuntimeException exception = thrown()
@@ -146,13 +154,13 @@ class MetricUnionAvailabilitySpec extends Specification {
         given:
         availability1.getAllAvailableIntervals() >> [
                 (metric1): [new Interval('2018-01-01/2018-02-01')],
-                ('column'): [new Interval('2018-01-01/2018-02-01')]
+                'column' : [new Interval('2018-01-01/2018-02-01')]
         ]
         availability2.getAllAvailableIntervals() >> [
                 (metric2): [new Interval('2019-01-01/2019-02-01')]
         ]
 
-        metricUnionAvailability = new MetricUnionAvailability(physicalTables, [metric1] as Set)
+        metricUnionAvailability = new MetricUnionAvailability(physicalTables, [metricColumn1] as Set)
 
         expect:
         metricUnionAvailability.getAllAvailableIntervals() == [
@@ -163,6 +171,36 @@ class MetricUnionAvailabilitySpec extends Specification {
 
     }
 
+    def "constructSubConstraint correctly intersect metrics configured on the table with the metrics supported by availability's underlying datasource"() {
+        given:
+        availability1.getAllAvailableIntervals() >> [
+                (metric1): []
+        ]
+        availability2.getAllAvailableIntervals() >> [
+                (metric2): [],
+                'ignored1': []
+        ]
+
+        metricUnionAvailability = new MetricUnionAvailability(physicalTables, [metricColumn1, metricColumn2] as Set)
+
+        DataSourceConstraint dataSourceConstraint = new DataSourceConstraint(
+                [] as Set,
+                [] as Set,
+                [] as Set,
+                [metric1, metric2, 'ignored2'] as Set,
+                [] as Set,
+                [] as Set,
+                [metric1, metric2, 'ignored2'] as Set,
+                [:]
+        )
+
+        PhysicalDataSourceConstraint physicalDataSourceConstraint = new PhysicalDataSourceConstraint(dataSourceConstraint, [metric1, metric2, 'ignored2'] as Set)
+
+        expect:
+        metricUnionAvailability.constructSubConstraint(physicalDataSourceConstraint).get(availability1).getMetricNames() == [metric1] as Set
+        metricUnionAvailability.constructSubConstraint(physicalDataSourceConstraint).get(availability2).getMetricNames() == [metric2] as Set
+    }
+
     @Unroll
     def "getAvailableIntervals returns the intersection of requested columns when available intervals have #reason"() {
         given:
@@ -173,32 +211,30 @@ class MetricUnionAvailabilitySpec extends Specification {
                 (metric2): []
         ]
 
-        availability1.getAvailableIntervals(_ as DataSourceConstraint) >> new SimplifiedIntervalList(
+        availability1.getAvailableIntervals(_ as PhysicalDataSourceConstraint) >> new SimplifiedIntervalList(
                 availableIntervals1.collect{it -> new Interval(it)} as Set
         )
-        availability2.getAvailableIntervals(_ as DataSourceConstraint) >> new SimplifiedIntervalList(
+        availability2.getAvailableIntervals(_ as PhysicalDataSourceConstraint) >> new SimplifiedIntervalList(
                 availableIntervals2.collect{it -> new Interval(it)} as Set
         )
 
-        metricUnionAvailability = new MetricUnionAvailability(physicalTables, [new MetricColumn(metric1), new MetricColumn(metric2)] as Set)
-
+        metricUnionAvailability = new MetricUnionAvailability(physicalTables, [metricColumn1, metricColumn2] as Set)
 
-
-        PhysicalDataSourceConstraint dataSourceConstraint = new PhysicalDataSourceConstraint(
-                [] as Set,
+        DataSourceConstraint dataSourceConstraint = new DataSourceConstraint(
                 [] as Set,
                 [] as Set,
-                ['metric1', 'metric2'] as Set,
                 [] as Set,
+                [metric1, metric2] as Set,
                 [] as Set,
                 [] as Set,
-                [:],
-                ['metric1', 'metric2'] as Set
+                [metric1, metric2] as Set,
+                [:]
         )
 
+        PhysicalDataSourceConstraint physicalDataSourceConstraint = new PhysicalDataSourceConstraint(dataSourceConstraint, [metric1, metric2] as Set)
+
         expect:
-        metricUnionAvailability.constructSubConstraint(dataSourceConstraint).size() == 2
-        metricUnionAvailability.getAvailableIntervals(dataSourceConstraint) == new SimplifiedIntervalList(
+        metricUnionAvailability.getAvailableIntervals(physicalDataSourceConstraint) == new SimplifiedIntervalList(
                 availableIntervals.collect{it -> new Interval(it)} as Set
         )
 

fili-core/src/test/groovy/com/yahoo/bard/webservice/table/resolver/DataSourceConstraintSpec.groovy

@@ -2,7 +2,6 @@
 // Licensed under the terms of the Apache license. Please see LICENSE.md file distributed with this work for terms.
 package com.yahoo.bard.webservice.table.resolver
 
-import com.yahoo.bard.webservice.data.metric.MetricColumn
 import spock.lang.Specification
 import spock.lang.Unroll
 

fili-core/src/test/groovy/com/yahoo/bard/webservice/table/resolver/PhysicalDataSourceConstraintSpec.groovy

@@ -16,10 +16,18 @@ class PhysicalDataSourceConstraintSpec extends Specification {
     PhysicalDataSourceConstraint physicalDataSourceConstraint
 
     def setup() {
-        dataSourceConstraint = Mock(DataSourceConstraint)
-        dataSourceConstraint
-        dataSourceConstraint.getAllColumnNames() >> (['columnOne', 'columnTwo', 'columnThree'] as Set)
-        physicalTableSchema = new PhysicalTableSchema(Mock(ZonedTimeGrain), [new Column('columnOne'), new Column('columnThree')], ['columnOne': 'column_one', 'columnTwo': 'column_two'])
+        dataSourceConstraint =  new DataSourceConstraint(
+                [] as Set,
+                [] as Set,
+                [] as Set,
+                ['columnThree', 'columnFour', 'columnFive'] as Set,
+                [] as Set,
+                [] as Set,
+                ['columnOne', 'columnTwo', 'columnThree', 'columnFour', 'columnFive'] as Set,
+                [:]
+        )
+
+        physicalTableSchema = new PhysicalTableSchema(Mock(ZonedTimeGrain), [new Column('columnOne'), new Column('columnThree'), new Column('columnFour'), new Column('columnFive')], ['columnOne': 'column_one', 'columnTwo': 'column_two'])
         physicalDataSourceConstraint = new PhysicalDataSourceConstraint(dataSourceConstraint, physicalTableSchema)
     }
 
@@ -30,6 +38,15 @@ class PhysicalDataSourceConstraintSpec extends Specification {
 
     def "physical data source constraint maps to physical name correctly even if mapping does not exist"() {
         expect:
-        physicalDataSourceConstraint.getAllColumnPhysicalNames() == ['column_one', 'columnThree'] as Set
+        physicalDataSourceConstraint.getAllColumnPhysicalNames() == ['column_one', 'columnThree', 'columnFour', 'columnFive'] as Set
+    }
+
+    def "withMetricIntersection correctly intersects metricNames and allPhysicalColumnNames with the provided metrics"() {
+        given:
+        PhysicalDataSourceConstraint subConstraint = physicalDataSourceConstraint.withMetricIntersection(['columnThree', 'columnFour'] as Set)
+
+        expect:
+        subConstraint.getMetricNames() == ['columnThree', 'columnFour'] as Set
+        subConstraint.withMetricIntersection(['columnThree', 'columnFour'] as Set).getAllColumnPhysicalNames() == ['column_one', 'columnThree', 'columnFour'] as Set
     }
 }