Mark a table if it is directly referenced by query

This allows us to identify the tables which are added as a part of query and tables which are referenced by views/row filter/column masks.

core/trino-main/src/main/java/io/trino/sql/analyzer/Analysis.java

@@ -570,7 +570,18 @@ public void registerTable(
             String authorization,
             Scope accessControlScope)
     {
-        tables.put(NodeRef.of(table), new TableEntry(handle, name, filters, columnMasks, authorization, accessControlScope));
+        tables.put(
+                NodeRef.of(table),
+                new TableEntry(
+                        handle,
+                        name,
+                        filters,
+                        columnMasks,
+                        authorization,
+                        accessControlScope,
+                        tablesForView.isEmpty() &&
+                                rowFilterScopes.isEmpty() &&
+                                columnMaskScopes.isEmpty()));
     }
 
     public ResolvedFunction getResolvedFunction(FunctionCall function)
@@ -941,7 +952,8 @@ public List<TableInfo> getReferencedTables()
                             rowFilters.getOrDefault(table, ImmutableList.of()).stream()
                                     .map(Expression::toString)
                                     .collect(toImmutableList()),
-                            columns);
+                            columns,
+                            info.isDirectlyReferenced());
                 })
                 .collect(toImmutableList());
     }
@@ -1419,21 +1431,24 @@ private static class TableEntry
         private final Map<Field, List<ViewExpression>> columnMasks;
         private final String authorization;
         private final Scope accessControlScope; // synthetic scope for analysis of row filters and masks
+        private final boolean directlyReferenced;
 
         public TableEntry(
                 Optional<TableHandle> handle,
                 QualifiedObjectName name,
                 List<ViewExpression> filters,
                 Map<Field, List<ViewExpression>> columnMasks,
                 String authorization,
-                Scope accessControlScope)
+                Scope accessControlScope,
+                boolean directlyReferenced)
         {
             this.handle = requireNonNull(handle, "handle is null");
             this.name = requireNonNull(name, "name is null");
             this.filters = requireNonNull(filters, "filters is null");
             this.columnMasks = requireNonNull(columnMasks, "columnMasks is null");
             this.authorization = requireNonNull(authorization, "authorization is null");
             this.accessControlScope = requireNonNull(accessControlScope, "accessControlScope is null");
+            this.directlyReferenced = directlyReferenced;
         }
 
         public Optional<TableHandle> getHandle()
@@ -1446,6 +1461,11 @@ public QualifiedObjectName getName()
             return name;
         }
 
+        public boolean isDirectlyReferenced()
+        {
+            return directlyReferenced;
+        }
+
         public List<ViewExpression> getFilters()
         {
             return filters;

core/trino-main/src/test/java/io/trino/connector/MockConnector.java

@@ -26,13 +26,16 @@
 import io.trino.spi.connector.ConnectorOutputTableHandle;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.ConnectorSplitManager;
+import io.trino.spi.connector.ConnectorSplitSource;
 import io.trino.spi.connector.ConnectorTableHandle;
 import io.trino.spi.connector.ConnectorTableMetadata;
 import io.trino.spi.connector.ConnectorTableProperties;
 import io.trino.spi.connector.ConnectorTransactionHandle;
 import io.trino.spi.connector.ConnectorViewDefinition;
 import io.trino.spi.connector.Constraint;
 import io.trino.spi.connector.ConstraintApplicationResult;
+import io.trino.spi.connector.DynamicFilter;
+import io.trino.spi.connector.FixedSplitSource;
 import io.trino.spi.connector.JoinApplicationResult;
 import io.trino.spi.connector.JoinCondition;
 import io.trino.spi.connector.JoinStatistics;
@@ -137,7 +140,13 @@ public ConnectorMetadata getMetadata(ConnectorTransactionHandle transaction)
     @Override
     public ConnectorSplitManager getSplitManager()
     {
-        return new ConnectorSplitManager() {};
+        return new ConnectorSplitManager() {
+            @Override
+            public ConnectorSplitSource getSplits(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorTableHandle table, SplitSchedulingStrategy splitSchedulingStrategy, DynamicFilter dynamicFilter)
+            {
+                return new FixedSplitSource(ImmutableList.of());
+            }
+        };
     }
 
     @Override

core/trino-main/src/test/java/io/trino/connector/MockConnectorAccessControl.java

@@ -19,9 +19,14 @@
 import io.trino.spi.security.ConnectorIdentity;
 import io.trino.spi.security.Privilege;
 import io.trino.spi.security.TrinoPrincipal;
+import io.trino.spi.security.ViewExpression;
+import io.trino.spi.type.Type;
 
 import java.util.Arrays;
+import java.util.Optional;
 import java.util.Set;
+import java.util.function.BiFunction;
+import java.util.function.Function;
 
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static io.trino.spi.security.AccessDeniedException.denyGrantSchemaPrivilege;
@@ -37,11 +42,19 @@ class MockConnectorAccessControl
 
     private final Grants<String> schemaGrants;
     private final Grants<SchemaTableName> tableGrants;
-
-    MockConnectorAccessControl(Grants<String> schemaGrants, Grants<SchemaTableName> tableGrants)
+    private final Function<SchemaTableName, ViewExpression> rowFilters;
+    private final BiFunction<SchemaTableName, String, ViewExpression> columnMasks;
+
+    MockConnectorAccessControl(
+            Grants<String> schemaGrants,
+            Grants<SchemaTableName> tableGrants,
+            Function<SchemaTableName, ViewExpression> rowFilters,
+            BiFunction<SchemaTableName, String, ViewExpression> columnMasks)
     {
         this.schemaGrants = requireNonNull(schemaGrants, "schemaGrants is null");
         this.tableGrants = requireNonNull(tableGrants, "tableGrants is null");
+        this.rowFilters = requireNonNull(rowFilters, "rowFilters is null");
+        this.columnMasks = requireNonNull(columnMasks, "columnMasks is null");
     }
 
     @Override
@@ -94,6 +107,18 @@ public void checkCanRevokeTablePrivilege(ConnectorSecurityContext context, Privi
         }
     }
 
+    @Override
+    public Optional<ViewExpression> getRowFilter(ConnectorSecurityContext context, SchemaTableName tableName)
+    {
+        return Optional.ofNullable(rowFilters.apply(tableName));
+    }
+
+    @Override
+    public Optional<ViewExpression> getColumnMask(ConnectorSecurityContext context, SchemaTableName tableName, String columnName, Type type)
+    {
+        return Optional.ofNullable(columnMasks.apply(tableName, columnName));
+    }
+
     public void grantSchemaPrivileges(String schemaName, Set<Privilege> privileges, TrinoPrincipal grantee, boolean grantOption)
     {
         schemaGrants.grant(grantee, schemaName, privileges, grantOption);

core/trino-main/src/test/java/io/trino/connector/MockConnectorFactory.java

@@ -44,6 +44,7 @@
 import io.trino.spi.eventlistener.EventListener;
 import io.trino.spi.expression.ConnectorExpression;
 import io.trino.spi.security.RoleGrant;
+import io.trino.spi.security.ViewExpression;
 
 import java.util.List;
 import java.util.Map;
@@ -241,6 +242,8 @@ public static final class Builder
         private Grants<SchemaTableName> tableGrants = new AllowAllGrants<>();
         private ApplyFilter applyFilter = (session, handle, constraint) -> Optional.empty();
         private ApplyTableScanRedirect applyTableScanRedirect = (session, handle) -> Optional.empty();
+        private Function<SchemaTableName, ViewExpression> rowFilter = (tableName) -> null;
+        private BiFunction<SchemaTableName, String, ViewExpression> columnMask = (tableName, columnName) -> null;
 
         public Builder withListSchemaNames(Function<ConnectorSession, List<String>> listSchemaNames)
         {
@@ -360,6 +363,18 @@ public Builder withTableGrants(Grants<SchemaTableName> tableGrants)
             return this;
         }
 
+        public Builder withRowFilter(Function<SchemaTableName, ViewExpression> rowFilter)
+        {
+            this.rowFilter = rowFilter;
+            return this;
+        }
+
+        public Builder withColumnMask(BiFunction<SchemaTableName, String, ViewExpression> columnMask)
+        {
+            this.columnMask = columnMask;
+            return this;
+        }
+
         public MockConnectorFactory build()
         {
             return new MockConnectorFactory(
@@ -379,7 +394,7 @@ public MockConnectorFactory build()
                     getTableProperties,
                     eventListeners,
                     roleGrants,
-                    new MockConnectorAccessControl(schemaGrants, tableGrants));
+                    new MockConnectorAccessControl(schemaGrants, tableGrants, rowFilter, columnMask));
         }
 
         public static Function<ConnectorSession, List<String>> defaultListSchemaNames()

core/trino-spi/src/main/java/io/trino/spi/eventlistener/TableInfo.java

@@ -28,15 +28,17 @@ public class TableInfo
 
     private final List<String> filters;
     private final List<ColumnInfo> columns;
+    private final boolean directlyReferenced;
 
-    public TableInfo(String catalog, String schema, String table, String authorization, List<String> filters, List<ColumnInfo> columns)
+    public TableInfo(String catalog, String schema, String table, String authorization, List<String> filters, List<ColumnInfo> columns, boolean directlyReferenced)
     {
         this.catalog = requireNonNull(catalog, "catalog is null");
         this.schema = requireNonNull(schema, "schema is null");
         this.table = requireNonNull(table, "table is null");
         this.authorization = requireNonNull(authorization, "authorization is null");
         this.filters = List.copyOf(requireNonNull(filters, "filters is null"));
         this.columns = List.copyOf(requireNonNull(columns, "columns is null"));
+        this.directlyReferenced = directlyReferenced;
     }
 
     @JsonProperty
@@ -74,4 +76,10 @@ public List<ColumnInfo> getColumns()
     {
         return columns;
     }
+
+    @JsonProperty
+    public boolean isDirectlyReferenced()
+    {
+        return directlyReferenced;
+    }
 }