feat: Add Flight SQL server support

engine/sql/src/main/java/io/deephaven/engine/sql/Sql.java

@@ -23,13 +23,15 @@
 import io.deephaven.sql.ScopeStaticImpl;
 import io.deephaven.sql.SqlAdapter;
 import io.deephaven.sql.TableInformation;
+import io.deephaven.util.annotations.InternalUseOnly;
 import io.deephaven.util.annotations.ScriptApi;
 
 import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.function.Function;
 
 /**
  * Experimental SQL execution. Subject to change.
@@ -47,35 +49,38 @@ public static TableSpec dryRun(String sql) {
         return dryRun(sql, currentScriptSessionNamedTables());
     }
 
+    @InternalUseOnly
+    public static TableSpec parseSql(String sql, Map<String, Table> scope, Function<String, TicketTable> ticketFunction,
+            Map<TicketTable, Table> out) {
+        return SqlAdapter.parseSql(sql, scope(scope, out, ticketFunction));
+    }
+
     private static Table evaluate(String sql, Map<String, Table> scope) {
         final Map<TicketTable, Table> map = new HashMap<>(scope.size());
-        final TableSpec tableSpec = parseSql(sql, scope, map);
+        final TableSpec tableSpec = parseSql(sql, scope, Sql::sqlref, map);
         log.debug().append("Executing. Graphviz representation:").nl().append(ToGraphvizDot.INSTANCE, tableSpec).endl();
         return tableSpec.logic().create(new TableCreatorTicketInterceptor(TableCreatorImpl.INSTANCE, map));
     }
 
     private static TableSpec dryRun(String sql, Map<String, Table> scope) {
-        final TableSpec tableSpec = parseSql(sql, scope, null);
+        final TableSpec tableSpec = parseSql(sql, scope, Sql::sqlref, null);
         log.info().append("Dry run. Graphviz representation:").nl().append(ToGraphvizDot.INSTANCE, tableSpec).endl();
         return tableSpec;
     }
 
-    private static TableSpec parseSql(String sql, Map<String, Table> scope, Map<TicketTable, Table> out) {
-        return SqlAdapter.parseSql(sql, scope(scope, out));
-    }
-
     private static TicketTable sqlref(String tableName) {
+        // The TicketTable can technically be anything unique (incrementing number, random, ...), but for
+        // visualization purposes it makes sense to use the (already unique) table name.
         return TicketTable.of(("sqlref/" + tableName).getBytes(StandardCharsets.UTF_8));
     }
 
-    private static Scope scope(Map<String, Table> scope, Map<TicketTable, Table> out) {
+    private static Scope scope(Map<String, Table> scope, Map<TicketTable, Table> out,
+            Function<String, TicketTable> ticketFunction) {
         final ScopeStaticImpl.Builder builder = ScopeStaticImpl.builder();
         for (Entry<String, Table> e : scope.entrySet()) {
             final String tableName = e.getKey();
             final Table table = e.getValue();
-            // The TicketTable can technically be anything unique (incrementing number, random, ...), but for
-            // visualization purposes it makes sense to use the (already unique) table name.
-            final TicketTable spec = sqlref(tableName);
+            final TicketTable spec = ticketFunction.apply(tableName);
             final List<String> qualifiedName = List.of(tableName);
             final TableHeader header = adapt(table.getDefinition());
             builder.addTables(TableInformation.of(qualifiedName, header, spec));

engine/sql/src/main/java/io/deephaven/engine/sql/TableCreatorTicketInterceptor.java

@@ -5,23 +5,17 @@
 
 import io.deephaven.engine.table.Table;
 import io.deephaven.qst.TableCreator;
-import io.deephaven.qst.table.EmptyTable;
-import io.deephaven.qst.table.InputTable;
-import io.deephaven.qst.table.MultiJoinInput;
-import io.deephaven.qst.table.NewTable;
+import io.deephaven.qst.TableCreatorDelegate;
 import io.deephaven.qst.table.TicketTable;
-import io.deephaven.qst.table.TimeTable;
 
-import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
-class TableCreatorTicketInterceptor implements TableCreator<Table> {
-    private final TableCreator<Table> delegate;
+class TableCreatorTicketInterceptor extends TableCreatorDelegate<Table> {
     private final Map<TicketTable, Table> map;
 
     public TableCreatorTicketInterceptor(TableCreator<Table> delegate, Map<TicketTable, Table> map) {
-        this.delegate = Objects.requireNonNull(delegate);
+        super(delegate);
         this.map = Objects.requireNonNull(map);
     }
 
@@ -31,36 +25,6 @@ public Table of(TicketTable ticketTable) {
         if (table != null) {
             return table;
         }
-        return delegate.of(ticketTable);
-    }
-
-    @Override
-    public Table of(NewTable newTable) {
-        return delegate.of(newTable);
-    }
-
-    @Override
-    public Table of(EmptyTable emptyTable) {
-        return delegate.of(emptyTable);
-    }
-
-    @Override
-    public Table of(TimeTable timeTable) {
-        return delegate.of(timeTable);
-    }
-
-    @Override
-    public Table of(InputTable inputTable) {
-        return delegate.of(inputTable);
-    }
-
-    @Override
-    public Table multiJoin(List<MultiJoinInput<Table>> multiJoinInputs) {
-        return delegate.multiJoin(multiJoinInputs);
-    }
-
-    @Override
-    public Table merge(Iterable<Table> tables) {
-        return delegate.merge(tables);
+        return super.of(ticketTable);
     }
 }

engine/table/src/main/java/io/deephaven/engine/util/TableTools.java

@@ -32,6 +32,7 @@
 import io.deephaven.util.annotations.ScriptApi;
 import io.deephaven.util.type.ArrayTypeUtils;
 import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
 
 import java.io.*;
 import java.nio.charset.StandardCharsets;
@@ -67,13 +68,13 @@ private static <T> BinaryOperator<T> throwingMerger() {
         };
     }
 
-    private static <T, K, U> Collector<T, ?, Map<K, U>> toLinkedMap(
+    private static <T, K, U> Collector<T, ?, LinkedHashMap<K, U>> toLinkedMap(
             Function<? super T, ? extends K> keyMapper,
             Function<? super T, ? extends U> valueMapper) {
         return Collectors.toMap(keyMapper, valueMapper, throwingMerger(), LinkedHashMap::new);
     }
 
-    private static final Collector<ColumnHolder<?>, ?, Map<String, ColumnSource<?>>> COLUMN_HOLDER_LINKEDMAP_COLLECTOR =
+    private static final Collector<ColumnHolder<?>, ?, LinkedHashMap<String, ColumnSource<?>>> COLUMN_HOLDER_LINKEDMAP_COLLECTOR =
             toLinkedMap(ColumnHolder::getName, ColumnHolder::getColumnSource);
 
     /////////// Utilities To Display Tables /////////////////
@@ -760,10 +761,16 @@ public static Table newTable(ColumnHolder<?>... columnHolders) {
     }
 
     public static Table newTable(TableDefinition definition, ColumnHolder<?>... columnHolders) {
+        return newTable(definition, null, columnHolders);
+    }
+
+    public static Table newTable(TableDefinition definition, @Nullable Map<String, Object> attributes,
+            ColumnHolder<?>... columnHolders) {
         checkSizes(columnHolders);
-        WritableRowSet rowSet = getRowSet(columnHolders);
-        Map<String, ColumnSource<?>> columns = Arrays.stream(columnHolders).collect(COLUMN_HOLDER_LINKEDMAP_COLLECTOR);
-        return new QueryTable(definition, rowSet.toTracking(), columns) {
+        final WritableRowSet rowSet = getRowSet(columnHolders);
+        final LinkedHashMap<String, ColumnSource<?>> columns =
+                Arrays.stream(columnHolders).collect(COLUMN_HOLDER_LINKEDMAP_COLLECTOR);
+        return new QueryTable(definition, rowSet.toTracking(), columns, null, attributes) {
             {
                 setFlat();
             }

extensions/barrage/src/main/java/io/deephaven/extensions/barrage/util/BarrageUtil.java

@@ -49,6 +49,7 @@
 import org.apache.arrow.util.Collections2;
 import org.apache.arrow.vector.types.TimeUnit;
 import org.apache.arrow.vector.types.Types;
+import org.apache.arrow.vector.types.Types.MinorType;
 import org.apache.arrow.vector.types.pojo.ArrowType;
 import org.apache.arrow.vector.types.pojo.Field;
 import org.apache.arrow.vector.types.pojo.FieldType;
@@ -736,6 +737,12 @@ private static ArrowType arrowTypeFor(Class<?> type) {
                 return Types.MinorType.FLOAT8.getType();
             case Object:
                 if (type.isArray()) {
+                    if (type == byte[].class) {
+                        return Types.MinorType.VARBINARY.getType();
+                    }
+                    // if (type == char[].class) {
+                    // return Types.MinorType.VARCHAR.getType();
+                    // }
                     return Types.MinorType.LIST.getType();
                 }
                 if (type == LocalDate.class) {

flightsql/README.md

@@ -0,0 +1,14 @@
+# FlightSQL
+
+## Client
+
+## JDBC
+
+The default FlightSQL JDBC driver uses cookie authorization; by default, this is not enabled on the Deephaven server.
+To enable this, the request header "x-deephaven-auth-cookie-request" must be set to "true".
+
+Example JDBC connection string to self-signed TLS:
+
+```
+jdbc:arrow-flight-sql://localhost:8443/?Authorization=Anonymous&useEncryption=1&disableCertificateVerification=1&x-deephaven-auth-cookie-request=true
+```

flightsql/build.gradle

@@ -0,0 +1,67 @@
+plugins {
+    id 'java-library'
+    id 'io.deephaven.project.register'
+}
+
+description = 'The Deephaven flight SQL library'
+
+sourceSets {
+    jdbcTest {
+        compileClasspath += sourceSets.test.output
+        runtimeClasspath += sourceSets.test.output
+    }
+}
+
+configurations {
+    jdbcTestImplementation.extendsFrom testImplementation
+    jdbcTestRuntimeOnly.extendsFrom testRuntimeOnly
+    jdbcTestAnnotationProcessor.extendsFrom testAnnotationProcessor
+}
+
+dependencies {
+    api project(':server')
+    implementation project(':sql')
+    implementation project(':engine-sql')
+    // :sql does not expose calcite as a dependency (maybe it should?); in the meantime, we want to make sure we can
+    // provide reasonable error messages to the client
+    implementation libs.calcite.core
+
+    implementation libs.dagger
+    implementation libs.arrow.flight.sql
+
+//    testImplementation project(':extensions-csv')
+    testImplementation project(':authorization')
+    testImplementation project(':server-test-utils')
+
+    testAnnotationProcessor libs.dagger.compiler
+    testImplementation libs.assertj
+    testImplementation platform(libs.junit.bom)
+    testImplementation libs.junit.jupiter
+    testRuntimeOnly libs.junit.platform.launcher
+    testRuntimeOnly libs.junit.vintage.engine
+    testRuntimeOnly project(':log-to-slf4j')
+    testRuntimeOnly libs.slf4j.simple
+
+    jdbcTestImplementation project(':server-jetty')
+    // Isolating to its own sourceSet / classpath because it breaks logging until we can upgrade to a newer version
+    // See https://github.com/apache/arrow/pull/40908
+    // See https://github.com/deephaven/deephaven-core/issues/5947
+    jdbcTestRuntimeOnly libs.arrow.flight.sql.jdbc
+}
+
+test {
+    useJUnitPlatform()
+}
+
+def jdbcTest = tasks.register('jdbcTest', Test) {
+    description = 'Runs JDBC tests.'
+    group = 'verification'
+
+    testClassesDirs = sourceSets.jdbcTest.output.classesDirs
+    classpath = sourceSets.jdbcTest.runtimeClasspath
+    shouldRunAfter test
+
+    useJUnitPlatform()
+}
+
+check.dependsOn jdbcTest

flightsql/gradle.properties

@@ -0,0 +1 @@
+io.deephaven.project.ProjectType=JAVA_PUBLIC

flightsql/src/jdbcTest/java/io/deephaven/server/DeephavenServerTestBase.java

@@ -0,0 +1,62 @@
+//
+// Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending
+//
+package io.deephaven.server;
+
+import io.deephaven.engine.context.ExecutionContext;
+import io.deephaven.io.logger.LogBuffer;
+import io.deephaven.io.logger.LogBufferGlobal;
+import io.deephaven.server.runner.GrpcServer;
+import io.deephaven.server.runner.MainHelper;
+import io.deephaven.util.SafeCloseable;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Timeout;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+@Timeout(30)
+public abstract class DeephavenServerTestBase {
+
+    public interface TestComponent {
+
+        GrpcServer server();
+
+        ExecutionContext executionContext();
+    }
+
+    protected TestComponent component;
+
+    private LogBuffer logBuffer;
+    private SafeCloseable executionContext;
+    private GrpcServer server;
+    protected int localPort;
+
+    protected abstract TestComponent component();
+
+    @BeforeAll
+    public static void setupOnce() throws IOException {
+        MainHelper.bootstrapProjectDirectories();
+    }
+
+    @BeforeEach
+    public void setup() throws IOException {
+        logBuffer = new LogBuffer(128);
+        LogBufferGlobal.setInstance(logBuffer);
+        component = component();
+        executionContext = component.executionContext().open();
+        server = component.server();
+        server.start();
+        localPort = server.getPort();
+    }
+
+    @AfterEach
+    void tearDown() throws InterruptedException {
+        server.stopWithTimeout(10, TimeUnit.SECONDS);
+        server.join();
+        executionContext.close();
+        LogBufferGlobal.clear(logBuffer);
+    }
+}