Add in concept of remnant-must-filter fields to handle the case of a …

…bypass-field being selected but not filtered on, a must-filter field not being selected (thus can no longer be defused), and filtering on the bypass field in the enclosing query, thus defusing the must-filter field

Fix style

core/src/main/java/org/apache/calcite/sql/validate/AbstractNamespace.java

@@ -28,7 +28,9 @@
 
 import org.checkerframework.checker.nullness.qual.Nullable;
 
+import java.util.Collections;
 import java.util.List;
+import java.util.Set;
 
 import static com.google.common.base.Preconditions.checkArgument;
 
@@ -64,6 +66,8 @@ abstract class AbstractNamespace implements SqlValidatorNamespace {
 
   protected ImmutableBitSet mustFilterBypassFields = ImmutableBitSet.of();
 
+  protected Set<SqlQualified> remnantMustFilterFields = Collections.emptySet();
+
   protected final @Nullable SqlNode enclosingNode;
 
   //~ Constructors -----------------------------------------------------------
@@ -175,6 +179,11 @@ abstract class AbstractNamespace implements SqlValidatorNamespace {
     return requireNonNull(mustFilterBypassFields,
         "mustFilterBypassFields (maybe validation is not complete?)");
   }
+
+  @Override public Set<SqlQualified> getRemnantMustFilterFields() {
+    return requireNonNull(remnantMustFilterFields,
+        "remnantMustFilterFields (maybe validation is not complete?");
+  }
   @Override public SqlMonotonicity getMonotonicity(String columnName) {
     return SqlMonotonicity.NOT_MONOTONIC;
   }

core/src/main/java/org/apache/calcite/sql/validate/SemanticTable.java

@@ -19,8 +19,10 @@
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 import java.util.List;
+import java.util.Set;
 
 import static java.util.Collections.emptyList;
+import static java.util.Collections.emptySet;
 
 /**
  * Extension to {@link SqlValidatorTable} with extra, optional metadata.
@@ -49,5 +51,20 @@ default boolean mustFilter(int column) {
     return getFilter(column) != null;
   }
 
-  default List<Integer> bypassFieldList() { return emptyList(); }
+  /**
+   * Returns a list of column ordinals (0-based) of fields that defuse
+   * must-filter columns when filtered on.
+   */
+  default List<Integer> bypassFieldList() {
+    return emptyList();
+  }
+
+  /**
+   * Returns a list of must-filter qualifieds that can only be defused by a selected bypass-field
+   * from the same table. The qualified was neither filtered on nor selected, and hence can no
+   * longer be defused by itself.
+   */
+  default Set<SqlQualified> remnantMustFilterFields() {
+    return emptySet();
+  }
 }

core/src/main/java/org/apache/calcite/sql/validate/SqlValidatorImpl.java

@@ -160,6 +160,7 @@
 import java.util.function.Supplier;
 import java.util.function.UnaryOperator;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
 import static com.google.common.base.Preconditions.checkArgument;
@@ -177,6 +178,7 @@
 import static org.apache.calcite.util.Util.first;
 
 import static java.util.Collections.emptyList;
+import static java.util.Collections.emptySet;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -1217,12 +1219,21 @@ protected void validateNamespace(final SqlValidatorNamespace namespace,
         // fields; these are neutralized if the consuming query filters on them.
         final ImmutableBitSet mustFilterFields =
             namespace.getMustFilterFields();
-        if (!mustFilterFields.isEmpty()) {
+        // Remnant must filter fields are fields that are not selected and cannot
+        // be defused unlessa bypass field defuses it.
+        // A top-level namespace must not have any remnant-must-filter fields.
+        final Set<SqlQualified> remnantMustFilterFields = namespace.getRemnantMustFilterFields();
+        if (!mustFilterFields.isEmpty() || !remnantMustFilterFields.isEmpty()) {
+          Stream<String> mustFilterStream =
+                  StreamSupport.stream(mustFilterFields.spliterator(), false)
+                      .map(namespace.getRowType().getFieldNames()::get);
+          Stream<String> remnantStream =
+                  StreamSupport.stream(remnantMustFilterFields.spliterator(), false)
+                      .map(q -> q.suffix().get(0));
+
           // Set of field names, sorted alphabetically for determinism.
-          Set<String> fieldNameSet =
-              StreamSupport.stream(mustFilterFields.spliterator(), false)
-                  .map(namespace.getRowType().getFieldNames()::get)
-                  .collect(Collectors.toCollection(TreeSet::new));
+          Set<String> fieldNameSet = Stream.concat(mustFilterStream, remnantStream)
+              .collect(Collectors.toCollection(TreeSet::new));
           throw newValidationError(node,
               RESOURCE.mustFilterFieldsMissing(fieldNameSet.toString()));
         }
@@ -4101,28 +4112,33 @@ protected void validateSelect(
     // Deduce which columns must be filtered.
     ns.mustFilterFields = ImmutableBitSet.of();
     ns.mustFilterBypassFields = ImmutableBitSet.of();
+    ns.remnantMustFilterFields = emptySet();
     if (from != null) {
+      final BitSet projectedNonFilteredBypassField = new BitSet();
       final Set<SqlQualified> qualifieds = new LinkedHashSet<>();
       final Set<SqlQualified> bypassQualifieds = new LinkedHashSet<>();
+      final Set<SqlQualified> remnantQualifieds = new LinkedHashSet<>();
       for (ScopeChild child : fromScope.children) {
         final List<String> fieldNames =
             child.namespace.getRowType().getFieldNames();
         toQualifieds(child.namespace.getMustFilterFields(), qualifieds, fromScope, child,
             fieldNames);
-        toQualifieds(child.namespace.getMustFilterBypassFields(), bypassQualifieds, fromScope, child, fieldNames);
+        toQualifieds(child.namespace.getMustFilterBypassFields(), bypassQualifieds, fromScope,
+            child, fieldNames);
+        remnantQualifieds.addAll(child.namespace.getRemnantMustFilterFields());
       }
-      if (!qualifieds.isEmpty()) {
+      if (!qualifieds.isEmpty() || !bypassQualifieds.isEmpty()) {
         if (select.getWhere() != null) {
           forEachQualified(select.getWhere(), getWhereScope(select),
               qualifieds::remove);
           purgeForBypassFields(select.getWhere(), getWhereScope(select),
-              qualifieds, bypassQualifieds);
+              qualifieds, bypassQualifieds, remnantQualifieds);
         }
         if (select.getHaving() != null) {
           forEachQualified(select.getHaving(), getHavingScope(select),
               qualifieds::remove);
           purgeForBypassFields(select.getHaving(), getHavingScope(select),
-              qualifieds, bypassQualifieds);
+              qualifieds, bypassQualifieds, remnantQualifieds);
         }
 
         // Each of the must-filter fields identified must be returned as a
@@ -4149,13 +4165,14 @@ protected void validateSelect(
               // SELECT item #i referenced a bypass column that was not filtered in the WHERE
               // or HAVING. It becomes a bypass column for our consumer.
               mustFilterBypassFields.set(i);
+              projectedNonFilteredBypassField.set(0);
             }
           }
         });
 
-        // If there are must-filter fields that are not in the SELECT clause,
-        // this is an error.
-        if (!qualifieds.isEmpty()) {
+        // If there are must-filter fields that are not in the SELECT clause and there were no
+        // bypass-fields on this table, this is an error.
+        if (!qualifieds.isEmpty() && !projectedNonFilteredBypassField.get(0)) {
           throw newValidationError(select,
               RESOURCE.mustFilterFieldsMissing(
                   qualifieds.stream()
@@ -4165,6 +4182,10 @@ protected void validateSelect(
         }
         ns.mustFilterFields = ImmutableBitSet.fromBitSet(mustFilterFields);
         ns.mustFilterBypassFields = ImmutableBitSet.fromBitSet(mustFilterBypassFields);
+        // Remaining must-filter fields can be defused by a bypass-field,
+        // so we pass this to the consumer.
+        ns.remnantMustFilterFields = Stream.of(remnantQualifieds, qualifieds)
+            .flatMap(Set::stream).collect(Collectors.toSet());
       }
     }
 
@@ -4197,10 +4218,12 @@ private static void forEachQualified(SqlNode node, SqlValidatorScope scope,
   }
 
   /* If the supplied SqlNode when fully qualified is in the set of bypassQualifieds, then we
-  remove all entries in the qualifieds set that belong to the same table identifier.
+  remove all entries in the qualifieds set as well as remnantMustFilterFields
+  that belong to the same table identifier.
    */
   private static void purgeForBypassFields(SqlNode node, SqlValidatorScope scope,
-      Set<SqlQualified> qualifieds, Set<SqlQualified> bypassQualifieds) {
+      Set<SqlQualified> qualifieds, Set<SqlQualified> bypassQualifieds,
+      Set<SqlQualified> remnantMustFilterFields) {
     node.accept(new SqlBasicVisitor<Void>() {
       @Override public Void visit(SqlIdentifier id) {
         final SqlQualified qualified = scope.fullyQualify(id);
@@ -4209,15 +4232,21 @@ private static void purgeForBypassFields(SqlNode node, SqlValidatorScope scope,
           Collection<SqlQualified> sameIdentifier = qualifieds.stream()
               .filter(q -> qualifiedMatchesIdentifier(q, qualified))
               .collect(Collectors.toList());
-          sameIdentifier.forEach (qualifieds::remove);
+          sameIdentifier.forEach(qualifieds::remove);
+
+          // Clear all the remnant must-filter qualifieds from the same table identifier
+          Collection<SqlQualified> sameIdentifier_ = remnantMustFilterFields.stream()
+              .filter(q -> qualifiedMatchesIdentifier(q, qualified))
+              .collect(Collectors.toList());
+          sameIdentifier_.forEach(remnantMustFilterFields::remove);
         }
         return null;
       }
     });
   }
 
   private static void toQualifieds(ImmutableBitSet fields, Set<SqlQualified> qualifiedSet,
-      SelectScope fromScope, ScopeChild child, List<String> fieldNames){
+      SelectScope fromScope, ScopeChild child, List<String> fieldNames) {
     fields.forEachInt(i ->
         qualifiedSet.add(
             SqlQualified.create(fromScope, 1, child.namespace,

core/src/main/java/org/apache/calcite/sql/validate/SqlValidatorNamespace.java

@@ -25,7 +25,9 @@
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.checkerframework.dataflow.qual.Pure;
 
+import java.util.Collections;
 import java.util.List;
+import java.util.Set;
 
 /**
  * A namespace describes the relation returned by a section of a SQL query.
@@ -223,4 +225,7 @@ default ImmutableBitSet getMustFilterFields() {
   default ImmutableBitSet getMustFilterBypassFields() {
     return ImmutableBitSet.of();
   }
+  default Set<SqlQualified> getRemnantMustFilterFields() {
+    return Collections.emptySet();
+  }
 }

core/src/main/java/org/apache/calcite/sql/validate/TableNamespace.java

@@ -34,6 +34,7 @@
 
 import org.checkerframework.checker.nullness.qual.Nullable;
 
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 
@@ -72,6 +73,9 @@ private TableNamespace(SqlValidatorImpl validator, SqlValidatorTable table,
         .ifPresent(semanticTable ->
             this.mustFilterBypassFields = semanticTable.bypassFieldList()
                 .stream().collect(toImmutableBitSet()));
+    table.maybeUnwrap(SemanticTable.class)
+        .ifPresent(semanticTable ->
+            this.remnantMustFilterFields = new HashSet<>(semanticTable.remnantMustFilterFields()));
 
     if (extendedFields.isEmpty()) {
       return table.getRowType();

core/src/test/java/org/apache/calcite/test/SqlValidatorTest.java

@@ -12450,15 +12450,14 @@ private void checkCustomColumnResolving(String table) {
             + "  select * from emp where empno = 1)\n"
             + "where job = 'doctor'")
         .ok();
-    fixture.withSql("select * from (\n"
-            + "  ^select ename from emp where empno = 1^)")
+    fixture.withSql("^select * from (\n"
+            + "  select ename from emp where empno = 1)^")
         .fails(missingFilters("JOB"));
 
-    // //This test is failing, needs new logic handling.
-    // fixture.withSql("select * from (\n"
-    //         + "  select ename from emp where empno = 1)"
-    //         + "where ename = '1'")
-    //     .ok();
+    fixture.withSql("select * from (\n"
+            + "  select job, ename from emp where empno = 1)"
+            + "where ename = '1'")
+        .ok();
 
     fixture.withSql("select * from (\n"
             + "  select empno, job from emp)\n"

testkit/src/main/java/org/apache/calcite/test/catalog/MockCatalogReader.java

@@ -79,6 +79,7 @@
 import org.apache.calcite.sql.validate.SqlMonotonicity;
 import org.apache.calcite.sql.validate.SqlNameMatcher;
 import org.apache.calcite.sql.validate.SqlNameMatchers;
+import org.apache.calcite.sql.validate.SqlQualified;
 import org.apache.calcite.sql.validate.SqlValidatorCatalogReader;
 import org.apache.calcite.sql.validate.SqlValidatorUtil;
 import org.apache.calcite.sql2rel.InitializerExpressionFactory;
@@ -102,6 +103,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -1028,6 +1030,8 @@ public static class MustFilterMockTable
     private final Map<String, String> fieldFilters;
     private final List<Integer> bypassFieldList;
 
+    private final Set<SqlQualified> remnantFieldFilters;
+
     MustFilterMockTable(MockCatalogReader catalogReader, String catalogName,
         String schemaName, String name, boolean stream, boolean temporal,
         double rowCount, @Nullable ColumnResolver resolver,
@@ -1037,6 +1041,7 @@ public static class MustFilterMockTable
           rowCount, resolver, initializerExpressionFactory);
       this.fieldFilters = ImmutableMap.copyOf(fieldFilters);
       this.bypassFieldList = ImmutableList.copyOf(bypassFieldList);
+      this.remnantFieldFilters = Collections.emptySet();
     }
 
     /** Creates a MustFilterMockTable. */
@@ -1066,6 +1071,10 @@ public static MustFilterMockTable create(MockCatalogReader catalogReader,
     @Override public List<Integer> bypassFieldList() {
       return bypassFieldList;
     }
+
+    @Override public Set<SqlQualified> remnantMustFilterFields() {
+      return remnantFieldFilters;
+    }
   }
 
   /** Wrapper around a {@link MockTable}, giving it a {@link Table} interface.

testkit/src/main/java/org/apache/calcite/test/catalog/MustFilterMockCatalogReader.java

@@ -48,6 +48,7 @@ public static SqlValidatorCatalogReader create(RelDataTypeFactory typeFactory,
     registerSchema(salesSchema);
 
     // Register "EMP" table. Must-filter fields are "EMPNO", "JOB".
+    // Bypass field of column (1): ENAME.
     MustFilterMockTable empTable =
         MustFilterMockTable.create(this, salesSchema, "EMP",
             false, 14, null, NullInitializerExpressionFactory.INSTANCE,
@@ -73,6 +74,7 @@ public static SqlValidatorCatalogReader create(RelDataTypeFactory typeFactory,
     registerTable(empTable);
 
     // Register "DEPT" table. "NAME" is a must-filter field.
+    // Bypass field of column (0): DEPTNO.
     MustFilterMockTable deptTable =
         MustFilterMockTable.create(this, salesSchema, "DEPT",
             false, 14, null, NullInitializerExpressionFactory.INSTANCE,