Updated to better handle authorized tables and use query scope resolver

extensions/flight-sql/src/main/java/io/deephaven/server/flightsql/FlightSqlResolver.java

@@ -33,7 +33,6 @@
 import io.deephaven.io.logger.Logger;
 import io.deephaven.proto.backplane.grpc.ExportNotification;
 import io.deephaven.proto.util.ByteHelper;
-import io.deephaven.qst.table.NewTable;
 import io.deephaven.qst.table.ParentsVisitor;
 import io.deephaven.qst.table.TableSpec;
 import io.deephaven.qst.table.TicketTable;
@@ -91,7 +90,6 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
 import java.security.SecureRandom;
 import java.time.Duration;
 import java.time.Instant;
@@ -102,7 +100,6 @@
 import java.util.Objects;
 import java.util.PrimitiveIterator;
 import java.util.Set;
-import java.util.concurrent.locks.Lock;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
@@ -650,9 +647,8 @@ private Table executeSqlQuery(SessionState session, String sql) {
         // See SQLTODO(catalog-reader-implementation)
         final ExecutionContext executionContext = ExecutionContext.getContext();
         final QueryScope queryScope = executionContext.getQueryScope();
-        // noinspection unchecked,rawtypes
         final Map<String, Table> queryScopeTables =
-                (Map<String, Table>) (Map) queryScope.toMap(queryScope::unwrapObject, (n, t) -> t instanceof Table);
+                queryScope.toMap(o -> queryScopeAuthorizedTableMapper(queryScope, o), (n, t) -> t != null);
         final TableSpec tableSpec = Sql.parseSql(sql, queryScopeTables, TicketTable::fromQueryScopeField, null);
 
         // We could consider doing finer-grained sharedLock in the future; right now, taking it for the whole operation
@@ -664,9 +660,10 @@ private Table executeSqlQuery(SessionState session, String sql) {
             if (!(node instanceof TicketTable)) {
                 continue;
             }
-            final String variableName = new String(((TicketTable) node).ticket(), StandardCharsets.UTF_8).substring(2);
-            final Table sourceTable = queryScopeTables.get(variableName);
-            Assert.neqNull(sourceTable, "sourceTable");
+            // We know that all TicketTables currently produced by Flight SQL will be global scope tickets
+            final Table sourceTable = scopeTicketResolver
+                    .<Table>resolve(session, ByteBuffer.wrap(((TicketTable) node).ticket()), "table")
+                    .get();
             if (sourceTable.isRefreshing()) {
                 hasRefreshingSource = true;
                 break;
@@ -690,6 +687,22 @@ private Table executeSqlQuery(SessionState session, String sql) {
         }
     }
 
+    private Table queryScopeTableMapper(QueryScope queryScope, Object object) {
+        if (object == null) {
+            return null;
+        }
+        object = queryScope.unwrapObject(object);
+        if (!(object instanceof Table)) {
+            return null;
+        }
+        return (Table) object;
+    }
+
+    private Table queryScopeAuthorizedTableMapper(QueryScope queryScope, Object object) {
+        final Table table = queryScopeTableMapper(queryScope, object);
+        return table == null ? null : authorization.transform(table);
+    }
+
     /**
      * This is the base class for "easy" commands; that is, commands that have a fixed schema and are cheap to
      * initialize.
@@ -1347,8 +1360,11 @@ private Table getTablesEmpty(boolean includeSchema, Map<String, Object> attribut
         private Table getTables(boolean includeSchema, QueryScope queryScope, Map<String, Object> attributes,
                 Predicate<String> tableNameFilter) {
             Objects.requireNonNull(attributes);
-            final Map<String, Table> queryScopeTables =
-                    (Map<String, Table>) (Map) queryScope.toMap(queryScope::unwrapObject, (n, t) -> t instanceof Table);
+            // Note: _not_ using queryScopeAuthorizedTable mapper; we can have a more efficient implementation when
+            // !includeSchema that only needs to check authorization.isDeniedAccess.
+            final Map<String, Table> queryScopeTables = queryScope.toMap(
+                    o -> queryScopeTableMapper(queryScope, o),
+                    (tableName, table) -> table != null && tableNameFilter.test(tableName));
             final int size = queryScopeTables.size();
             final String[] catalogNames = new String[size];
             final String[] dbSchemaNames = new String[size];
@@ -1358,9 +1374,6 @@ private Table getTables(boolean includeSchema, QueryScope queryScope, Map<String
             int count = 0;
             for (Entry<String, Table> e : queryScopeTables.entrySet()) {
                 final String tableName = e.getKey();
-                if (!tableNameFilter.test(tableName)) {
-                    continue;
-                }
                 final Schema schema;
                 if (includeSchema) {
                     final Table table = authorization.transform(e.getValue());