Adds InputTableService support for blink tables (deephaven#4934)

This adds `TablePublisher#inputTable`, which adds a very simple `InputTableUpdater` implementation to the blink table.

Also, some further simplifications in the spirit of deephaven#4923: `InputTableUpdater#getDescription`, `InputTableUpdater#getTable`, and `InputTableUpdater#canEdit` were removed due to no usages.

The documentation on `InputTableUpdater` was updated to make it more generally useful in other contexts, such as here with blink tables.

The `InputTableService` implementation is now taking advantage of the async methods on `InputTableUpdater`.

Fixes deephaven#4915

engine/table/src/main/java/io/deephaven/engine/table/impl/TableCreatorImpl.java

@@ -12,6 +12,7 @@
 import io.deephaven.engine.table.impl.util.KeyedArrayBackedInputTable;
 import io.deephaven.engine.util.TableTools;
 import io.deephaven.qst.TableCreator;
+import io.deephaven.qst.table.BlinkInputTable;
 import io.deephaven.qst.table.EmptyTable;
 import io.deephaven.qst.table.InMemoryAppendOnlyInputTable;
 import io.deephaven.qst.table.InMemoryKeyBackedInputTable;
@@ -24,8 +25,10 @@
 import io.deephaven.qst.table.Clock;
 import io.deephaven.qst.table.ClockSystem;
 import io.deephaven.qst.table.TimeTable;
+import io.deephaven.stream.TablePublisher;
 
 import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
@@ -78,8 +81,8 @@ public final Table of(TicketTable ticketTable) {
     }
 
     @Override
-    public final UpdatableTable of(InputTable inputTable) {
-        return UpdatableTableAdapter.of(inputTable);
+    public final Table of(InputTable inputTable) {
+        return InputTableAdapter.of(inputTable);
     }
 
 
@@ -153,10 +156,12 @@ public io.deephaven.base.clock.Clock visit(ClockSystem system) {
         }
     }
 
-    enum UpdatableTableAdapter implements InputTable.Visitor<UpdatableTable> {
+    enum InputTableAdapter implements InputTable.Visitor<Table> {
         INSTANCE;
 
-        public static UpdatableTable of(InputTable inputTable) {
+        private static final AtomicInteger blinkTableCount = new AtomicInteger();
+
+        public static Table of(InputTable inputTable) {
             return inputTable.walk(INSTANCE);
         }
 
@@ -172,6 +177,15 @@ public UpdatableTable visit(InMemoryKeyBackedInputTable inMemoryKeyBacked) {
             final String[] keyColumnNames = inMemoryKeyBacked.keys().toArray(String[]::new);
             return KeyedArrayBackedInputTable.make(definition, keyColumnNames);
         }
+
+        @Override
+        public Table visit(BlinkInputTable blinkInputTable) {
+            final TableDefinition definition = DefinitionAdapter.of(blinkInputTable.schema());
+            return TablePublisher
+                    .of(TableCreatorImpl.class.getSimpleName() + ".BLINK-" + blinkTableCount.getAndIncrement(),
+                            definition, null, null)
+                    .inputTable();
+        }
     }
 
     enum DefinitionAdapter implements TableSchema.Visitor<TableDefinition> {

engine/table/src/main/java/io/deephaven/engine/table/impl/util/BaseArrayBackedInputTable.java

@@ -303,11 +303,6 @@ private void checkAsyncEditSafety(@NotNull final Table changeData) {
             }
         }
 
-        @Override
-        public String getDescription() {
-            return description;
-        }
-
         void waitForSequence(long sequence) {
             if (updateGraph.exclusiveLock().isHeldByCurrentThread()) {
                 // We're holding the lock. currentTable had better be refreshing. Wait on its UGP condition
@@ -346,16 +341,5 @@ private Map<String, WritableColumnSource<Object>> buildSourcesMap(int capacity,
             return sources;
         }
 
-        @Override
-        public Table getTable() {
-            return BaseArrayBackedInputTable.this;
-        }
-
-        @Override
-        public boolean canEdit() {
-            // TODO: Should we be more restrictive, or provide a mechanism for determining which users can edit this
-            // table beyond "they have a handle to it"?
-            return true;
-        }
     }
 }

engine/table/src/main/java/io/deephaven/engine/util/input/InputTableUpdater.java

@@ -91,9 +91,14 @@ default void validateDelete(Table tableToDelete) {
     /**
      * Write {@code newData} to this table. Added rows with keys that match existing rows will instead replace those
      * rows, if supported.
+     *
+     * <p>
+     * This method will block until the add is "completed", where the definition of "completed" is implementation
+     * dependenent.
+     *
      * <p>
-     * This method will block until the rows are added. As a result, this method is not suitable for use from a
-     * {@link io.deephaven.engine.table.TableListener table listener} or any other
+     * For implementations where "completed" means "visible in the next update graph cycle", this method is not suitable
+     * for use from a {@link io.deephaven.engine.table.TableListener table listener} or any other
      * {@link io.deephaven.engine.updategraph.NotificationQueue.Notification notification}-dispatched callback
      * dispatched by this InputTable's {@link io.deephaven.engine.updategraph.UpdateGraph update graph}. It may be
      * suitable to delete from another update graph if doing so does not introduce any cycles.
@@ -106,6 +111,11 @@ default void validateDelete(Table tableToDelete) {
     /**
      * Write {@code newData} to this table. Added rows with keys that match existing rows replace those rows, if
      * supported.
+     *
+     * <p>
+     * The callback to {@code listener} will happen when the add has "completed", where the definition of "completed" is
+     * implementation dependenent. It's possible that the callback happens immediately on the same thread.
+     *
      * <p>
      * This method will <em>not</em> block, and can be safely used from a {@link io.deephaven.engine.table.TableListener
      * table listener} or any other {@link io.deephaven.engine.updategraph.NotificationQueue.Notification
@@ -120,9 +130,14 @@ default void validateDelete(Table tableToDelete) {
 
     /**
      * Delete the keys contained in {@code table} from this input table.
+     *
+     * <p>
+     * This method will block until the delete is "completed", where the definition of "completed" is implementation
+     * dependenent.
+     *
      * <p>
-     * This method will block until the rows are deleted. As a result, this method is not suitable for use from a
-     * {@link io.deephaven.engine.table.TableListener table listener} or any other
+     * For implementations where "completed" means "visible in the next update graph cycle", this method is not suitable
+     * for use from a {@link io.deephaven.engine.table.TableListener table listener} or any other
      * {@link io.deephaven.engine.updategraph.NotificationQueue.Notification notification}-dispatched callback
      * dispatched by this InputTable's {@link io.deephaven.engine.updategraph.UpdateGraph update graph}. It may be
      * suitable to delete from another update graph if doing so does not introduce any cycles.
@@ -137,6 +152,11 @@ default void delete(Table table) throws IOException {
 
     /**
      * Delete the keys contained in table from this input table.
+     *
+     * <p>
+     * The callback to {@code listener} will happen when the delete has "completed", where the definition of "completed"
+     * is implementation dependenent. It's possible that the callback happens immediately on the same thread.
+     * 
      * <p>
      * This method will <em>not</em> block, and can be safely used from a {@link io.deephaven.engine.table.TableListener
      * table listener} or any other {@link io.deephaven.engine.updategraph.NotificationQueue.Notification
@@ -151,20 +171,6 @@ default void deleteAsync(Table table, InputTableStatusListener listener) {
         throw new UnsupportedOperationException("Table does not support deletes");
     }
 
-    /**
-     * Return a user-readable description of this InputTable.
-     *
-     * @return a description of this input table
-     */
-    String getDescription();
-
-    /**
-     * Returns a Deephaven table that contains the current data for this InputTable.
-     *
-     * @return the current data in this InputTable.
-     */
-    Table getTable();
-
     /**
      * Returns true if the specified column is a key.
      *
@@ -184,12 +190,4 @@ default boolean isKey(String columnName) {
     default boolean hasColumn(String columnName) {
         return getTableDefinition().getColumnNames().contains(columnName);
     }
-
-    /**
-     * Queries whether this InputTable is editable in the current context.
-     *
-     * @return true if this InputTable may be edited, false otherwise TODO (deephaven/deephaven-core/issues/255): Add
-     *         AuthContext and whatever else is appropriate
-     */
-    boolean canEdit();
 }

engine/table/src/main/java/io/deephaven/stream/TablePublisher.java

@@ -5,6 +5,8 @@
 import io.deephaven.engine.table.TableDefinition;
 import io.deephaven.engine.table.impl.sources.ArrayBackedColumnSource;
 import io.deephaven.engine.updategraph.UpdateGraph;
+import io.deephaven.engine.util.input.InputTableStatusListener;
+import io.deephaven.engine.util.input.InputTableUpdater;
 import io.deephaven.util.annotations.TestUseOnly;
 
 import javax.annotation.Nullable;
@@ -169,6 +171,29 @@ public boolean isAlive() {
         return adapter.isAlive();
     }
 
+    /**
+     * Creates a new {@link Table#BLINK_TABLE_ATTRIBUTE blink table} with its {@link Table#getAttribute(String)
+     * attribute} {@value Table#INPUT_TABLE_ATTRIBUTE} set to an {@link InputTableUpdater} implementation based on
+     * {@code this}. The implementation's definition of "completed" with respect to {@link InputTableUpdater#add(Table)}
+     * and {@link InputTableUpdater#addAsync(Table, InputTableStatusListener)} matches the semantics provided by
+     * {@link #add(Table)} - that is, "completed" means that a snapshot of {@code newData} has been taken and handed
+     * off. The implementation does not implement {@link InputTableUpdater#delete(Table)} nor
+     * {@link InputTableUpdater#deleteAsync(Table, InputTableStatusListener)}.
+     *
+     * <p>
+     * May return {@code null} if invoked more than once and the initial caller does not enforce strong reachability of
+     * the result.
+     *
+     * @return the input-table blink table
+     */
+    public Table inputTable() {
+        final Table table = adapter.table();
+        if (table == null) {
+            return null;
+        }
+        return table.withAttributes(Map.of(Table.INPUT_TABLE_ATTRIBUTE, publisher.inputTableUpdater()));
+    }
+
     @TestUseOnly
     void runForUnitTests() {
         adapter.run();

engine/table/src/main/java/io/deephaven/stream/TableStreamPublisherImpl.java

@@ -17,11 +17,14 @@
 import io.deephaven.engine.table.impl.remote.ConstructSnapshot.SnapshotFunction;
 import io.deephaven.engine.table.impl.remote.ConstructSnapshot.State;
 import io.deephaven.engine.table.impl.sources.ReinterpretUtils;
+import io.deephaven.engine.util.input.InputTableStatusListener;
+import io.deephaven.engine.util.input.InputTableUpdater;
 import io.deephaven.util.SafeCloseable;
 import io.deephaven.util.SafeCloseableArray;
 import org.jetbrains.annotations.NotNull;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.stream.Stream;
@@ -94,6 +97,39 @@ public void shutdown() {
         }
     }
 
+    public InputTableUpdater inputTableUpdater() {
+        return new InputTableAdapter();
+    }
+
+    private class InputTableAdapter implements InputTableUpdater {
+        @Override
+        public TableDefinition getTableDefinition() {
+            return definition;
+        }
+
+        @Override
+        public void add(Table newData) {
+            TableStreamPublisherImpl.this.add(newData);
+        }
+
+        @Override
+        public void addAsync(Table newData, InputTableStatusListener listener) {
+            try {
+                TableStreamPublisherImpl.this.add(newData);
+            } catch (Throwable t) {
+                listener.onError(t);
+                return;
+            }
+            listener.onSuccess();
+        }
+
+        @Override
+        public List<String> getKeyNames() {
+            return Collections.emptyList();
+        }
+
+    }
+
     private class FillChunks implements SnapshotFunction {
         private final Table table;
         private final ColumnSource<?>[] sources;

java-client/session/src/main/java/io/deephaven/client/impl/BatchTableRequestBuilder.java

@@ -40,6 +40,7 @@
 import io.deephaven.proto.backplane.grpc.Condition;
 import io.deephaven.proto.backplane.grpc.CreateInputTableRequest;
 import io.deephaven.proto.backplane.grpc.CreateInputTableRequest.InputTableKind;
+import io.deephaven.proto.backplane.grpc.CreateInputTableRequest.InputTableKind.Blink;
 import io.deephaven.proto.backplane.grpc.CreateInputTableRequest.InputTableKind.InMemoryAppendOnly;
 import io.deephaven.proto.backplane.grpc.CreateInputTableRequest.InputTableKind.InMemoryKeyBacked;
 import io.deephaven.proto.backplane.grpc.CrossJoinTablesRequest;
@@ -76,6 +77,7 @@
 import io.deephaven.qst.table.AggregateAllTable;
 import io.deephaven.qst.table.AggregateTable;
 import io.deephaven.qst.table.AsOfJoinTable;
+import io.deephaven.qst.table.BlinkInputTable;
 import io.deephaven.qst.table.Clock.Visitor;
 import io.deephaven.qst.table.ClockSystem;
 import io.deephaven.qst.table.DropColumnsTable;
@@ -535,6 +537,11 @@ public InputTableKind visit(InMemoryKeyBackedInputTable inMemoryKeyBacked) {
                     return InputTableKind.newBuilder().setInMemoryKeyBacked(
                             InMemoryKeyBacked.newBuilder().addAllKeyColumns(inMemoryKeyBacked.keys())).build();
                 }
+
+                @Override
+                public InputTableKind visit(BlinkInputTable blinkInputTable) {
+                    return InputTableKind.newBuilder().setBlink(Blink.getDefaultInstance()).build();
+                }
             }));
             return op(Builder::setCreateInputTable, builder);
         }

proto/proto-backplane-grpc/src/main/proto/deephaven/proto/table.proto

@@ -1210,9 +1210,12 @@ message CreateInputTableRequest {
     message InMemoryKeyBacked {
       repeated string key_columns = 1;
     }
+    message Blink {
+    }
     oneof kind {
       InMemoryAppendOnly in_memory_append_only = 1;
       InMemoryKeyBacked in_memory_key_backed = 2;
+      Blink blink = 3;
     }
   }
 

qst/src/main/java/io/deephaven/qst/table/BlinkInputTable.java

@@ -0,0 +1,33 @@
+/**
+ * Copyright (c) 2016-2022 Deephaven Data Labs and Patent Pending
+ */
+package io.deephaven.qst.table;
+
+import io.deephaven.annotations.NodeStyle;
+import org.immutables.value.Value.Immutable;
+import org.immutables.value.Value.Parameter;
+
+import java.util.UUID;
+
+/**
+ * Creates a blink input-table.
+ */
+@Immutable
+@NodeStyle
+public abstract class BlinkInputTable extends InputTableBase {
+
+    public static BlinkInputTable of(TableSchema schema) {
+        return ImmutableBlinkInputTable.of(schema, UUID.randomUUID());
+    }
+
+    @Parameter
+    public abstract TableSchema schema();
+
+    @Parameter
+    abstract UUID id();
+
+    @Override
+    public final <R> R walk(InputTable.Visitor<R> visitor) {
+        return visitor.visit(this);
+    }
+}

qst/src/main/java/io/deephaven/qst/table/InputTable.java

@@ -25,5 +25,7 @@ interface Visitor<R> {
         R visit(InMemoryAppendOnlyInputTable inMemoryAppendOnly);
 
         R visit(InMemoryKeyBackedInputTable inMemoryKeyBacked);
+
+        R visit(BlinkInputTable blinkInputTable);
     }
 }

qst/src/main/java/io/deephaven/qst/table/TableLabelVisitor.java

@@ -156,6 +156,11 @@ public String visit(InMemoryAppendOnlyInputTable inMemoryAppendOnly) {
             public String visit(InMemoryKeyBackedInputTable inMemoryKeyBacked) {
                 return "InMemoryKeyBackedInputTable(...)";
             }
+
+            @Override
+            public String visit(BlinkInputTable blinkInputTable) {
+                return "BlinkInputTable(...)";
+            }
         });
     }