diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/TableCreatorImpl.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/TableCreatorImpl.java index fe745db0b2a..76e643d4dcd 100644 --- a/engine/table/src/main/java/io/deephaven/engine/table/impl/TableCreatorImpl.java +++ b/engine/table/src/main/java/io/deephaven/engine/table/impl/TableCreatorImpl.java @@ -8,8 +8,8 @@ import io.deephaven.engine.table.Table; import io.deephaven.engine.table.TableDefinition; import io.deephaven.engine.table.TableFactory; -import io.deephaven.engine.table.impl.util.AppendOnlyArrayBackedMutableTable; -import io.deephaven.engine.table.impl.util.KeyedArrayBackedMutableTable; +import io.deephaven.engine.table.impl.util.AppendOnlyArrayBackedInputTable; +import io.deephaven.engine.table.impl.util.KeyedArrayBackedInputTable; import io.deephaven.engine.util.TableTools; import io.deephaven.qst.TableCreator; import io.deephaven.qst.table.EmptyTable; @@ -163,14 +163,14 @@ public static UpdatableTable of(InputTable inputTable) { @Override public UpdatableTable visit(InMemoryAppendOnlyInputTable inMemoryAppendOnly) { final TableDefinition definition = DefinitionAdapter.of(inMemoryAppendOnly.schema()); - return AppendOnlyArrayBackedMutableTable.make(definition); + return AppendOnlyArrayBackedInputTable.make(definition); } @Override public UpdatableTable visit(InMemoryKeyBackedInputTable inMemoryKeyBacked) { final TableDefinition definition = DefinitionAdapter.of(inMemoryKeyBacked.schema()); final String[] keyColumnNames = inMemoryKeyBacked.keys().toArray(String[]::new); - return KeyedArrayBackedMutableTable.make(definition, keyColumnNames); + return KeyedArrayBackedInputTable.make(definition, keyColumnNames); } } diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/util/AppendOnlyArrayBackedMutableTable.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/util/AppendOnlyArrayBackedInputTable.java similarity index 59% rename from engine/table/src/main/java/io/deephaven/engine/table/impl/util/AppendOnlyArrayBackedMutableTable.java rename to engine/table/src/main/java/io/deephaven/engine/table/impl/util/AppendOnlyArrayBackedInputTable.java index f40908ed679..a65210dc3ea 100644 --- a/engine/table/src/main/java/io/deephaven/engine/table/impl/util/AppendOnlyArrayBackedMutableTable.java +++ b/engine/table/src/main/java/io/deephaven/engine/table/impl/util/AppendOnlyArrayBackedInputTable.java @@ -9,7 +9,6 @@ import io.deephaven.engine.rowset.RowSet; import io.deephaven.engine.rowset.RowSetFactory; import io.deephaven.engine.rowset.RowSequenceFactory; -import io.deephaven.engine.util.config.InputTableStatusListener; import io.deephaven.engine.table.impl.QueryTable; import io.deephaven.engine.table.impl.sources.NullValueColumnSource; import io.deephaven.engine.table.ChunkSink; @@ -18,15 +17,13 @@ import java.util.Collections; import java.util.List; -import java.util.Map; -import java.util.function.Consumer; /** * An in-memory table that allows you to add rows as if it were an InputTable, which can be updated on the UGP. *

* The table is not keyed, all rows are added to the end of the table. Deletions and edits are not permitted. */ -public class AppendOnlyArrayBackedMutableTable extends BaseArrayBackedMutableTable { +public class AppendOnlyArrayBackedInputTable extends BaseArrayBackedInputTable { static final String DEFAULT_DESCRIPTION = "Append Only In-Memory Input Table"; /** @@ -36,64 +33,40 @@ public class AppendOnlyArrayBackedMutableTable extends BaseArrayBackedMutableTab * * @return an empty AppendOnlyArrayBackedMutableTable with the given definition */ - public static AppendOnlyArrayBackedMutableTable make(@NotNull TableDefinition definition) { - return make(definition, Collections.emptyMap()); - } - - /** - * Create an empty AppendOnlyArrayBackedMutableTable with the given definition. - * - * @param definition the definition of the new table. - * @param enumValues a map of column names to enumeration values - * - * @return an empty AppendOnlyArrayBackedMutableTable with the given definition - */ - public static AppendOnlyArrayBackedMutableTable make(@NotNull TableDefinition definition, - final Map enumValues) { + public static AppendOnlyArrayBackedInputTable make( + @NotNull TableDefinition definition) { // noinspection resource return make(new QueryTable(definition, RowSetFactory.empty().toTracking(), - NullValueColumnSource.createColumnSourceMap(definition)), enumValues); - } - - /** - * Create an AppendOnlyArrayBackedMutableTable with the given initial data. - * - * @param initialTable the initial values to copy into the AppendOnlyArrayBackedMutableTable - * - * @return an empty AppendOnlyArrayBackedMutableTable with the given definition - */ - public static AppendOnlyArrayBackedMutableTable make(final Table initialTable) { - return make(initialTable, Collections.emptyMap()); + NullValueColumnSource.createColumnSourceMap(definition))); } /** * Create an AppendOnlyArrayBackedMutableTable with the given initial data. * * @param initialTable the initial values to copy into the AppendOnlyArrayBackedMutableTable - * @param enumValues a map of column names to enumeration values * * @return an empty AppendOnlyArrayBackedMutableTable with the given definition */ - public static AppendOnlyArrayBackedMutableTable make(final Table initialTable, - final Map enumValues) { - final AppendOnlyArrayBackedMutableTable result = new AppendOnlyArrayBackedMutableTable( - initialTable.getDefinition(), enumValues, new ProcessPendingUpdater()); + public static AppendOnlyArrayBackedInputTable make(final Table initialTable) { + final AppendOnlyArrayBackedInputTable result = + new AppendOnlyArrayBackedInputTable( + initialTable.getDefinition(), new ProcessPendingUpdater()); result.setAttribute(Table.ADD_ONLY_TABLE_ATTRIBUTE, Boolean.TRUE); + result.setAttribute(Table.APPEND_ONLY_TABLE_ATTRIBUTE, Boolean.TRUE); result.setFlat(); processInitial(initialTable, result); return result; } - private AppendOnlyArrayBackedMutableTable(@NotNull TableDefinition definition, - final Map enumValues, final ProcessPendingUpdater processPendingUpdater) { + private AppendOnlyArrayBackedInputTable(@NotNull TableDefinition definition, + final ProcessPendingUpdater processPendingUpdater) { // noinspection resource super(RowSetFactory.empty().toTracking(), makeColumnSourceMap(definition), - enumValues, processPendingUpdater); + processPendingUpdater); } @Override - protected void processPendingTable(Table table, boolean allowEdits, RowSetChangeRecorder rowSetChangeRecorder, - Consumer errorNotifier) { + protected void processPendingTable(Table table, RowSetChangeRecorder rowSetChangeRecorder) { try (final RowSet addRowSet = table.getRowSet().copy()) { final long firstRow = nextRow; final long lastRow = firstRow + addRowSet.intSize() - 1; @@ -135,28 +108,15 @@ protected List getKeyNames() { } @Override - ArrayBackedMutableInputTable makeHandler() { - return new AppendOnlyArrayBackedMutableInputTable(); + ArrayBackedInputTableUpdater makeUpdater() { + return new Updater(); } - private class AppendOnlyArrayBackedMutableInputTable extends ArrayBackedMutableInputTable { - @Override - public void setRows(@NotNull Table defaultValues, int[] rowArray, Map[] valueArray, - InputTableStatusListener listener) { - throw new UnsupportedOperationException(); - } + private class Updater extends ArrayBackedInputTableUpdater { @Override public void validateDelete(Table tableToDelete) { throw new UnsupportedOperationException("Table doesn't support delete operation"); } - - @Override - public void addRows(Map[] valueArray, boolean allowEdits, InputTableStatusListener listener) { - if (allowEdits) { - throw new UnsupportedOperationException(); - } - super.addRows(valueArray, allowEdits, listener); - } } } diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/util/BaseArrayBackedMutableTable.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/util/BaseArrayBackedInputTable.java similarity index 62% rename from engine/table/src/main/java/io/deephaven/engine/table/impl/util/BaseArrayBackedMutableTable.java rename to engine/table/src/main/java/io/deephaven/engine/table/impl/util/BaseArrayBackedInputTable.java index fc1c75d69df..f74f4b82907 100644 --- a/engine/table/src/main/java/io/deephaven/engine/table/impl/util/BaseArrayBackedMutableTable.java +++ b/engine/table/src/main/java/io/deephaven/engine/table/impl/util/BaseArrayBackedInputTable.java @@ -4,9 +4,6 @@ package io.deephaven.engine.table.impl.util; import io.deephaven.base.verify.Assert; -import io.deephaven.base.verify.Require; -import io.deephaven.datastructures.util.CollectionUtil; -import io.deephaven.engine.rowset.RowSet; import io.deephaven.engine.rowset.RowSetBuilderSequential; import io.deephaven.engine.rowset.RowSetFactory; import io.deephaven.engine.rowset.TrackingRowSet; @@ -15,9 +12,8 @@ import io.deephaven.engine.table.TableDefinition; import io.deephaven.engine.table.WritableColumnSource; import io.deephaven.engine.table.impl.sources.ArrayBackedColumnSource; -import io.deephaven.engine.util.config.InputTableStatusListener; -import io.deephaven.engine.util.config.MutableInputTable; -import io.deephaven.engine.table.impl.QueryTable; +import io.deephaven.engine.util.input.InputTableStatusListener; +import io.deephaven.engine.util.input.InputTableUpdater; import io.deephaven.engine.table.impl.UpdatableTable; import io.deephaven.engine.table.ColumnSource; import io.deephaven.util.annotations.TestUseOnly; @@ -26,11 +22,8 @@ import java.io.IOException; import java.util.*; import java.util.concurrent.CompletableFuture; -import java.util.function.Consumer; -abstract class BaseArrayBackedMutableTable extends UpdatableTable { - - private static final Object[] BOOLEAN_ENUM_ARRAY = new Object[] {true, false, null}; +abstract class BaseArrayBackedInputTable extends UpdatableTable { /** * Queue of pending changes. Only synchronized access is permitted. @@ -45,30 +38,27 @@ abstract class BaseArrayBackedMutableTable extends UpdatableTable { */ private long processedSequence = 0L; - private final Map enumValues; - private String description = getDefaultDescription(); private Runnable onPendingChange = updateGraph::requestRefresh; long nextRow = 0; private long pendingProcessed = -1L; - public BaseArrayBackedMutableTable(TrackingRowSet rowSet, Map> nameToColumnSource, - Map enumValues, ProcessPendingUpdater processPendingUpdater) { + public BaseArrayBackedInputTable(TrackingRowSet rowSet, Map> nameToColumnSource, + ProcessPendingUpdater processPendingUpdater) { super(rowSet, nameToColumnSource, processPendingUpdater); - this.enumValues = enumValues; - MutableInputTable mutableInputTable = makeHandler(); - setAttribute(Table.INPUT_TABLE_ATTRIBUTE, mutableInputTable); + InputTableUpdater inputTableUpdater = makeUpdater(); + setAttribute(Table.INPUT_TABLE_ATTRIBUTE, inputTableUpdater); setRefreshing(true); processPendingUpdater.setThis(this); } - public MutableInputTable mutableInputTable() { - return (MutableInputTable) getAttribute(Table.INPUT_TABLE_ATTRIBUTE); + public InputTableUpdater inputTable() { + return (InputTableUpdater) getAttribute(Table.INPUT_TABLE_ATTRIBUTE); } public Table readOnlyCopy() { - return copy(BaseArrayBackedMutableTable::applicableForReadOnly); + return copy(BaseArrayBackedInputTable::applicableForReadOnly); } private static boolean applicableForReadOnly(String attributeName) { @@ -84,9 +74,9 @@ private static boolean applicableForReadOnly(String attributeName) { return resultMap; } - static void processInitial(Table initialTable, BaseArrayBackedMutableTable result) { + static void processInitial(Table initialTable, BaseArrayBackedInputTable result) { final RowSetBuilderSequential builder = RowSetFactory.builderSequential(); - result.processPendingTable(initialTable, true, new RowSetChangeRecorder() { + result.processPendingTable(initialTable, new RowSetChangeRecorder() { @Override public void addRowKey(long key) { builder.appendKey(key); @@ -101,14 +91,13 @@ public void removeRowKey(long key) { public void modifyRowKey(long key) { throw new UnsupportedOperationException(); } - }, (e) -> { }); result.getRowSet().writableCast().insert(builder.build()); result.getRowSet().writableCast().initializePreviousValue(); result.getUpdateGraph().addSource(result); } - public BaseArrayBackedMutableTable setDescription(String newDescription) { + public BaseArrayBackedInputTable setDescription(String newDescription) { this.description = newDescription; return this; } @@ -132,8 +121,7 @@ private void processPending(RowSetChangeRecorder rowSetChangeRecorder) { if (pendingChange.delete) { processPendingDelete(pendingChange.table, rowSetChangeRecorder); } else { - processPendingTable(pendingChange.table, pendingChange.allowEdits, rowSetChangeRecorder, - (e) -> pendingChange.error = e); + processPendingTable(pendingChange.table, rowSetChangeRecorder); } pendingProcessed = pendingChange.sequence; } @@ -154,8 +142,7 @@ public void run() { } } - protected abstract void processPendingTable(Table table, boolean allowEdits, - RowSetChangeRecorder rowSetChangeRecorder, Consumer errorNotifier); + protected abstract void processPendingTable(Table table, RowSetChangeRecorder rowSetChangeRecorder); protected abstract void processPendingDelete(Table table, RowSetChangeRecorder rowSetChangeRecorder); @@ -164,74 +151,73 @@ protected abstract void processPendingTable(Table table, boolean allowEdits, protected abstract List getKeyNames(); protected static class ProcessPendingUpdater implements Updater { - private BaseArrayBackedMutableTable baseArrayBackedMutableTable; + private BaseArrayBackedInputTable baseArrayBackedInputTable; @Override public void accept(RowSetChangeRecorder rowSetChangeRecorder) { - baseArrayBackedMutableTable.processPending(rowSetChangeRecorder); + baseArrayBackedInputTable.processPending(rowSetChangeRecorder); } - public void setThis(BaseArrayBackedMutableTable keyedArrayBackedMutableTable) { - this.baseArrayBackedMutableTable = keyedArrayBackedMutableTable; + public void setThis(BaseArrayBackedInputTable keyedArrayBackedMutableTable) { + this.baseArrayBackedInputTable = keyedArrayBackedMutableTable; } } private final class PendingChange { final boolean delete; + @NotNull final Table table; final long sequence; - final boolean allowEdits; String error; - private PendingChange(Table table, boolean delete, boolean allowEdits) { + private PendingChange(@NotNull Table table, boolean delete) { Assert.holdsLock(pendingChanges, "pendingChanges"); + Assert.neqNull(table, "table"); this.table = table; this.delete = delete; - this.allowEdits = allowEdits; this.sequence = ++enqueuedSequence; } } - ArrayBackedMutableInputTable makeHandler() { - return new ArrayBackedMutableInputTable(); + ArrayBackedInputTableUpdater makeUpdater() { + return new ArrayBackedInputTableUpdater(); } - protected class ArrayBackedMutableInputTable implements MutableInputTable { + protected class ArrayBackedInputTableUpdater implements InputTableUpdater { @Override public List getKeyNames() { - return BaseArrayBackedMutableTable.this.getKeyNames(); + return BaseArrayBackedInputTable.this.getKeyNames(); } @Override public TableDefinition getTableDefinition() { - return BaseArrayBackedMutableTable.this.getDefinition(); + return BaseArrayBackedInputTable.this.getDefinition(); } @Override public void add(@NotNull final Table newData) throws IOException { checkBlockingEditSafety(); - PendingChange pendingChange = enqueueAddition(newData, true); + PendingChange pendingChange = enqueueAddition(newData); blockingContinuation(pendingChange); } @Override public void addAsync( @NotNull final Table newData, - final boolean allowEdits, @NotNull final InputTableStatusListener listener) { checkAsyncEditSafety(newData); - final PendingChange pendingChange = enqueueAddition(newData, allowEdits); + final PendingChange pendingChange = enqueueAddition(newData); asynchronousContinuation(pendingChange, listener); } - private PendingChange enqueueAddition(@NotNull final Table newData, final boolean allowEdits) { + private PendingChange enqueueAddition(@NotNull final Table newData) { validateAddOrModify(newData); // we want to get a clean copy of the table; that can not change out from under us or result in long reads // during our UGP run final Table newDataSnapshot = snapshotData(newData); final PendingChange pendingChange; synchronized (pendingChanges) { - pendingChange = new PendingChange(newDataSnapshot, false, allowEdits); + pendingChange = new PendingChange(newDataSnapshot, false); pendingChanges.add(pendingChange); } onPendingChange.run(); @@ -239,38 +225,33 @@ private PendingChange enqueueAddition(@NotNull final Table newData, final boolea } @Override - public void delete(@NotNull final Table table, @NotNull final TrackingRowSet rowsToDelete) throws IOException { + public void delete(@NotNull final Table table) throws IOException { checkBlockingEditSafety(); - final PendingChange pendingChange = enqueueDeletion(table, rowsToDelete); + final PendingChange pendingChange = enqueueDeletion(table); blockingContinuation(pendingChange); } @Override public void deleteAsync( @NotNull final Table table, - @NotNull final TrackingRowSet rowsToDelete, @NotNull final InputTableStatusListener listener) { checkAsyncEditSafety(table); - final PendingChange pendingChange = enqueueDeletion(table, rowsToDelete); + final PendingChange pendingChange = enqueueDeletion(table); asynchronousContinuation(pendingChange, listener); } - private PendingChange enqueueDeletion(@NotNull final Table table, @NotNull final TrackingRowSet rowsToDelete) { + private PendingChange enqueueDeletion(@NotNull final Table table) { validateDelete(table); - final Table oldDataSnapshot = snapshotData(table, rowsToDelete); + final Table oldDataSnapshot = snapshotData(table); final PendingChange pendingChange; synchronized (pendingChanges) { - pendingChange = new PendingChange(oldDataSnapshot, true, false); + pendingChange = new PendingChange(oldDataSnapshot, true); pendingChanges.add(pendingChange); } onPendingChange.run(); return pendingChange; } - private Table snapshotData(@NotNull final Table data, @NotNull final TrackingRowSet rowSet) { - return snapshotData(data.getSubTable(rowSet)); - } - private Table snapshotData(@NotNull final Table data) { Table dataSnapshot; if (data.isRefreshing()) { @@ -333,7 +314,7 @@ void waitForSequence(long sequence) { // in order to allow updates. while (processedSequence < sequence) { try { - BaseArrayBackedMutableTable.this.awaitUpdate(); + BaseArrayBackedInputTable.this.awaitUpdate(); } catch (InterruptedException ignored) { } } @@ -350,84 +331,6 @@ void waitForSequence(long sequence) { } } - @Override - public void setRows(@NotNull Table defaultValues, int[] rowArray, Map[] valueArray, - InputTableStatusListener listener) { - Assert.neqNull(defaultValues, "defaultValues"); - if (defaultValues.isRefreshing()) { - updateGraph.checkInitiateSerialTableOperation(); - } - - final List> columnDefinitions = getTableDefinition().getColumns(); - final Map> sources = - buildSourcesMap(valueArray.length, columnDefinitions); - final String[] kabmtColumns = - getTableDefinition().getColumnNames().toArray(CollectionUtil.ZERO_LENGTH_STRING_ARRAY); - // noinspection unchecked - final WritableColumnSource[] sourcesByPosition = - Arrays.stream(kabmtColumns).map(sources::get).toArray(WritableColumnSource[]::new); - - final Set missingColumns = new HashSet<>(getTableDefinition().getColumnNames()); - - for (final Map.Entry> entry : defaultValues.getColumnSourceMap() - .entrySet()) { - final String colName = entry.getKey(); - if (!sources.containsKey(colName)) { - continue; - } - final ColumnSource cs = Require.neqNull(entry.getValue(), "defaultValue column source: " + colName); - final WritableColumnSource dest = - Require.neqNull(sources.get(colName), "destination column source: " + colName); - - final RowSet defaultValuesRowSet = defaultValues.getRowSet(); - for (int rr = 0; rr < rowArray.length; ++rr) { - final long key = defaultValuesRowSet.get(rowArray[rr]); - dest.set(rr, cs.get(key)); - } - - missingColumns.remove(colName); - } - - for (int ii = 0; ii < valueArray.length; ++ii) { - final Map passedInValues = valueArray[ii]; - - for (int cc = 0; cc < sourcesByPosition.length; cc++) { - final String colName = kabmtColumns[cc]; - if (passedInValues.containsKey(colName)) { - sourcesByPosition[cc].set(ii, passedInValues.get(colName)); - } else if (missingColumns.contains(colName)) { - throw new IllegalArgumentException("No value specified for " + colName + " row " + ii); - } - } - } - - // noinspection resource - final QueryTable newData = new QueryTable(getTableDefinition(), - RowSetFactory.flat(valueArray.length).toTracking(), sources); - addAsync(newData, true, listener); - } - - @Override - public void addRows(Map[] valueArray, boolean allowEdits, InputTableStatusListener listener) { - final List> columnDefinitions = getTableDefinition().getColumns(); - final Map> sources = - buildSourcesMap(valueArray.length, columnDefinitions); - - for (int rowNumber = 0; rowNumber < valueArray.length; rowNumber++) { - final Map values = valueArray[rowNumber]; - for (final ColumnDefinition columnDefinition : columnDefinitions) { - sources.get(columnDefinition.getName()).set(rowNumber, values.get(columnDefinition.getName())); - } - - } - - // noinspection resource - final QueryTable newData = new QueryTable(getTableDefinition(), - RowSetFactory.flat(valueArray.length).toTracking(), sources); - - addAsync(newData, allowEdits, listener); - } - @NotNull private Map> buildSourcesMap(int capacity, List> columnDefinitions) { @@ -443,17 +346,9 @@ private Map> buildSourcesMap(int capacity, return sources; } - @Override - public Object[] getEnumsForColumn(String columnName) { - if (getTableDefinition().getColumn(columnName).getDataType().equals(Boolean.class)) { - return BOOLEAN_ENUM_ARRAY; - } - return enumValues.get(columnName); - } - @Override public Table getTable() { - return BaseArrayBackedMutableTable.this; + return BaseArrayBackedInputTable.this; } @Override diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/util/KeyedArrayBackedMutableTable.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/util/KeyedArrayBackedInputTable.java similarity index 76% rename from engine/table/src/main/java/io/deephaven/engine/table/impl/util/KeyedArrayBackedMutableTable.java rename to engine/table/src/main/java/io/deephaven/engine/table/impl/util/KeyedArrayBackedInputTable.java index ad4221bbb90..1eaeba52a01 100644 --- a/engine/table/src/main/java/io/deephaven/engine/table/impl/util/KeyedArrayBackedMutableTable.java +++ b/engine/table/src/main/java/io/deephaven/engine/table/impl/util/KeyedArrayBackedInputTable.java @@ -20,14 +20,13 @@ import org.jetbrains.annotations.NotNull; import java.util.*; -import java.util.function.Consumer; /** * An in-memory table that has keys for each row, which can be updated on the UGP. *

* This is used to implement in-memory editable table columns from web plugins. */ -public class KeyedArrayBackedMutableTable extends BaseArrayBackedMutableTable { +public class KeyedArrayBackedInputTable extends BaseArrayBackedInputTable { private static final String DEFAULT_DESCRIPTION = "In-Memory Input Table"; @@ -47,44 +46,13 @@ public class KeyedArrayBackedMutableTable extends BaseArrayBackedMutableTable { * * @return an empty KeyedArrayBackedMutableTable with the given definition and key columns */ - public static KeyedArrayBackedMutableTable make(@NotNull TableDefinition definition, + public static KeyedArrayBackedInputTable make(@NotNull TableDefinition definition, final String... keyColumnNames) { // noinspection resource return make(new QueryTable(definition, RowSetFactory.empty().toTracking(), NullValueColumnSource.createColumnSourceMap(definition)), keyColumnNames); } - /** - * Create an empty KeyedArrayBackedMutableTable. - * - * @param definition the definition of the table to create - * @param enumValues a map of column names to enumeration values - * @param keyColumnNames the name of the key columns - * - * @return an empty KeyedArrayBackedMutableTable with the given definition and key columns - */ - public static KeyedArrayBackedMutableTable make(@NotNull TableDefinition definition, - final Map enumValues, final String... keyColumnNames) { - // noinspection resource - return make(new QueryTable(definition, RowSetFactory.empty().toTracking(), - NullValueColumnSource.createColumnSourceMap(definition)), enumValues, keyColumnNames); - } - - /** - * Create an empty KeyedArrayBackedMutableTable. - *

- * The initialTable is processed in order, so if there are duplicate keys only the last row is reflected in the - * output. - * - * @param initialTable the initial values to copy into the KeyedArrayBackedMutableTable - * @param keyColumnNames the name of the key columns - * - * @return an empty KeyedArrayBackedMutableTable with the given definition and key columns - */ - public static KeyedArrayBackedMutableTable make(final Table initialTable, final String... keyColumnNames) { - return make(initialTable, Collections.emptyMap(), keyColumnNames); - } - /** * Create an empty KeyedArrayBackedMutableTable. *

@@ -92,25 +60,23 @@ public static KeyedArrayBackedMutableTable make(final Table initialTable, final * output. * * @param initialTable the initial values to copy into the KeyedArrayBackedMutableTable - * @param enumValues a map of column names to enumeration values * @param keyColumnNames the name of the key columns * * @return an empty KeyedArrayBackedMutableTable with the given definition and key columns */ - public static KeyedArrayBackedMutableTable make(final Table initialTable, final Map enumValues, - final String... keyColumnNames) { - final KeyedArrayBackedMutableTable result = new KeyedArrayBackedMutableTable(initialTable.getDefinition(), - keyColumnNames, enumValues, new ProcessPendingUpdater()); + public static KeyedArrayBackedInputTable make(final Table initialTable, final String... keyColumnNames) { + final KeyedArrayBackedInputTable result = new KeyedArrayBackedInputTable(initialTable.getDefinition(), + keyColumnNames, new ProcessPendingUpdater()); processInitial(initialTable, result); result.startTrackingPrev(); return result; } - private KeyedArrayBackedMutableTable(@NotNull TableDefinition definition, final String[] keyColumnNames, - final Map enumValues, final ProcessPendingUpdater processPendingUpdater) { + private KeyedArrayBackedInputTable(@NotNull TableDefinition definition, final String[] keyColumnNames, + final ProcessPendingUpdater processPendingUpdater) { // noinspection resource super(RowSetFactory.empty().toTracking(), makeColumnSourceMap(definition), - enumValues, processPendingUpdater); + processPendingUpdater); final List missingKeyColumns = new ArrayList<>(Arrays.asList(keyColumnNames)); missingKeyColumns.removeAll(definition.getColumnNames()); if (!missingKeyColumns.isEmpty()) { @@ -135,13 +101,11 @@ private void startTrackingPrev() { } @Override - protected void processPendingTable(Table table, boolean allowEdits, RowSetChangeRecorder rowSetChangeRecorder, - Consumer errorNotifier) { + protected void processPendingTable(Table table, RowSetChangeRecorder rowSetChangeRecorder) { final ChunkSource keySource = makeKeySource(table); final int chunkCapacity = table.intSize(); long rowToInsert = nextRow; - final StringBuilder errorBuilder = new StringBuilder(); try (final RowSet addRowSet = table.getRowSet().copy(); final WritableLongChunk destinations = WritableLongChunk.makeWritableChunk(chunkCapacity); @@ -161,25 +125,13 @@ protected void processPendingTable(Table table, boolean allowEdits, RowSetChange keyToRowMap.put(key, rowNumber); rowSetChangeRecorder.addRowKey(rowNumber); destinations.set(ii, rowNumber); - } else if (allowEdits) { + } else { rowSetChangeRecorder.modifyRowKey(rowNumber); destinations.set(ii, rowNumber); - } else { - // invalid edit - if (errorBuilder.length() > 0) { - errorBuilder.append(", ").append(key); - } else { - errorBuilder.append("Can not edit keys ").append(key); - } } } } - if (errorBuilder.length() > 0) { - errorNotifier.accept(errorBuilder.toString()); - return; - } - for (long ii = nextRow; ii < rowToInsert; ++ii) { rowSetChangeRecorder.addRowKey(ii); } diff --git a/engine/table/src/main/java/io/deephaven/engine/util/config/InputTableEnumGetter.java b/engine/table/src/main/java/io/deephaven/engine/util/config/InputTableEnumGetter.java deleted file mode 100644 index d861e125377..00000000000 --- a/engine/table/src/main/java/io/deephaven/engine/util/config/InputTableEnumGetter.java +++ /dev/null @@ -1,11 +0,0 @@ -/** - * Copyright (c) 2016-2022 Deephaven Data Labs and Patent Pending - */ -package io.deephaven.engine.util.config; - -/** - * Accessor interface for enumeration constants for an input table column. - */ -public interface InputTableEnumGetter { - Object[] getEnumsForColumn(String columnName); -} diff --git a/engine/table/src/main/java/io/deephaven/engine/util/config/InputTableRowSetter.java b/engine/table/src/main/java/io/deephaven/engine/util/config/InputTableRowSetter.java deleted file mode 100644 index 1d058ea6567..00000000000 --- a/engine/table/src/main/java/io/deephaven/engine/util/config/InputTableRowSetter.java +++ /dev/null @@ -1,94 +0,0 @@ -/** - * Copyright (c) 2016-2022 Deephaven Data Labs and Patent Pending - */ -package io.deephaven.engine.util.config; - -import io.deephaven.engine.table.Table; - -import java.util.Map; - -public interface InputTableRowSetter { - /** - * Set the values of the column specified by the input, filling in missing data using the parameter 'table' as the - * previous value source. This method will be invoked asynchronously. Users may use - * {@link #setRows(Table, int[], Map[], InputTableStatusListener)} to be notified of asynchronous results. - * - * @param table The table to use as the previous value source - * @param row The row key to set - * @param values A map of column name to value to set. - */ - default void setRow(Table table, int row, Map values) { - // noinspection unchecked - setRows(table, new int[] {row}, new Map[] {values}); - } - - /** - * Set the values of the columns specified by the input, filling in missing data using the parameter 'table' as the - * previous value source. This method will be invoked asynchronously. Users may use - * {@link #setRows(Table, int[], Map[], InputTableStatusListener)} to be notified of asynchronous results. - * - * @param table The table to use as the previous value source - * @param rowArray The row keys to update. - * @param valueArray The new values. - */ - default void setRows(Table table, int[] rowArray, Map[] valueArray) { - setRows(table, rowArray, valueArray, InputTableStatusListener.DEFAULT); - } - - /** - * Set the values of the columns specified by the input, filling in missing data using the parameter 'table' as the - * previous value source. This method will be invoked asynchronously. The input listener will be notified on - * success/failure - * - * @param table The table to use as the previous value source - * @param rowArray The row keys to update. - * @param valueArray The new values. - * @param listener The listener to notify on asynchronous results. - */ - void setRows(Table table, int[] rowArray, Map[] valueArray, InputTableStatusListener listener); - - /** - * Add the specified row to the table. Duplicate keys will be overwritten. This method will execute asynchronously. - * Users may use {@link #addRow(Map, boolean, InputTableStatusListener)} to handle the result of the asynchronous - * write. - * - * @param values The values to write. - */ - default void addRow(Map values) { - // noinspection unchecked - addRows(new Map[] {values}); - } - - /** - * Add the specified rows to the table. Duplicate keys will be overwritten. This method will execute asynchronously. - * Users may use {@link #addRows(Map[], boolean, InputTableStatusListener)} to handle the asynchronous result. - * - * @param valueArray The values to write. - */ - default void addRows(Map[] valueArray) { - addRows(valueArray, true, InputTableStatusListener.DEFAULT); - } - - /** - * Add the specified row to the table, optionally overwriting existing keys. This method will execute - * asynchronously, the input listener will be notified on success/failure. - * - * @param valueArray The value to write. - * @param allowEdits Should pre-existing keys be overwritten? - * @param listener The listener to report asynchronous result to. - */ - default void addRow(Map valueArray, boolean allowEdits, InputTableStatusListener listener) { - // noinspection unchecked - addRows(new Map[] {valueArray}, allowEdits, listener); - } - - /** - * Add the specified rows to the table, optionally overwriting existing keys. This method will execute - * asynchronously, the input listener will be notified on success/failure. - * - * @param valueArray The values to write. - * @param allowEdits Should pre-existing keys be overwritten? - * @param listener The listener to report asynchronous results to. - */ - void addRows(Map[] valueArray, boolean allowEdits, InputTableStatusListener listener); -} diff --git a/engine/table/src/main/java/io/deephaven/engine/util/config/InputTableStatusListener.java b/engine/table/src/main/java/io/deephaven/engine/util/input/InputTableStatusListener.java similarity index 92% rename from engine/table/src/main/java/io/deephaven/engine/util/config/InputTableStatusListener.java rename to engine/table/src/main/java/io/deephaven/engine/util/input/InputTableStatusListener.java index 8061f253642..2676d20a11f 100644 --- a/engine/table/src/main/java/io/deephaven/engine/util/config/InputTableStatusListener.java +++ b/engine/table/src/main/java/io/deephaven/engine/util/input/InputTableStatusListener.java @@ -1,7 +1,7 @@ /** * Copyright (c) 2016-2022 Deephaven Data Labs and Patent Pending */ -package io.deephaven.engine.util.config; +package io.deephaven.engine.util.input; import io.deephaven.internal.log.LoggerFactory; import io.deephaven.io.logger.Logger; @@ -37,7 +37,7 @@ public void onSuccess() { } /** - * Handle an error that occured during an input table write. + * Handle an error that occurred during an input table write. * * @param t the error. */ diff --git a/engine/table/src/main/java/io/deephaven/engine/util/config/MutableInputTable.java b/engine/table/src/main/java/io/deephaven/engine/util/input/InputTableUpdater.java similarity index 71% rename from engine/table/src/main/java/io/deephaven/engine/util/config/MutableInputTable.java rename to engine/table/src/main/java/io/deephaven/engine/util/input/InputTableUpdater.java index 202256ca7ea..e43053d37ef 100644 --- a/engine/table/src/main/java/io/deephaven/engine/util/config/MutableInputTable.java +++ b/engine/table/src/main/java/io/deephaven/engine/util/input/InputTableUpdater.java @@ -1,13 +1,12 @@ /** * Copyright (c) 2016-2022 Deephaven Data Labs and Patent Pending */ -package io.deephaven.engine.util.config; +package io.deephaven.engine.util.input; import io.deephaven.engine.exceptions.ArgumentException; import io.deephaven.engine.table.ColumnDefinition; import io.deephaven.engine.table.Table; import io.deephaven.engine.table.TableDefinition; -import io.deephaven.engine.rowset.TrackingRowSet; import java.io.IOException; import java.util.List; @@ -15,12 +14,12 @@ /** * A minimal interface for mutable shared tables, providing the ability to write to the table instance this is attached - * to. MutableInputTable instances are set on the table as an attribute. + * to. InputTable instances are set on the table as an attribute. *

* Implementations of this interface will make their own guarantees about how atomically changes will be applied and * what operations they support. */ -public interface MutableInputTable extends InputTableRowSetter, InputTableEnumGetter { +public interface InputTableUpdater { /** * Gets the names of the key columns. @@ -85,7 +84,7 @@ default void validateDelete(Table tableToDelete) { error.append("Unknown key columns: ").append(extraKeys); } if (error.length() > 0) { - throw new ArgumentException("Invalid Key Table Definition: " + error.toString()); + throw new ArgumentException("Invalid Key Table Definition: " + error); } } @@ -96,8 +95,8 @@ default void validateDelete(Table tableToDelete) { * 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 * {@link io.deephaven.engine.updategraph.NotificationQueue.Notification notification}-dispatched callback - * dispatched by this MutableInputTable'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. + * 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. * * @param newData The data to write to this table * @throws IOException If there is an error writing the data @@ -105,8 +104,8 @@ default void validateDelete(Table tableToDelete) { void add(Table newData) throws IOException; /** - * Write {@code newData} to this table. Added rows with keys that match existing rows will instead replace those - * rows, if supported and {@code allowEdits == true}. + * Write {@code newData} to this table. Added rows with keys that match existing rows replace those rows, if + * supported. *

* This method will not 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 @@ -115,11 +114,9 @@ default void validateDelete(Table tableToDelete) { * cycle. * * @param newData The data to write to this table - * @param allowEdits Whether added rows with keys that match existing rows will instead replace those rows, or - * result in an error * @param listener The listener for asynchronous results */ - void addAsync(Table newData, boolean allowEdits, InputTableStatusListener listener); + void addAsync(Table newData, InputTableStatusListener listener); /** * Delete the keys contained in {@code table} from this input table. @@ -127,37 +124,19 @@ default void validateDelete(Table tableToDelete) { * 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 * {@link io.deephaven.engine.updategraph.NotificationQueue.Notification notification}-dispatched callback - * dispatched by this MutableInputTable'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. + * 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. * * @param table The rows to delete * @throws IOException If a problem occurred while deleting the rows. * @throws UnsupportedOperationException If this table does not support deletes */ default void delete(Table table) throws IOException { - delete(table, table.getRowSet()); - } - - /** - * Delete the keys contained in {@code table.subTable(rowSet)} from this input table. - *

- * 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 - * {@link io.deephaven.engine.updategraph.NotificationQueue.Notification notification}-dispatched callback - * dispatched by this MutableInputTable'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. - * - * @param table Table containing the rows to delete - * @param rowSet The rows to delete - * @throws IOException If a problem occurred while deleting the rows - * @throws UnsupportedOperationException If this table does not support deletes - */ - default void delete(Table table, TrackingRowSet rowSet) throws IOException { throw new UnsupportedOperationException("Table does not support deletes"); } /** - * Delete the keys contained in {@code table.subTable(rowSet)} from this input table. + * Delete the keys contained in table from this input table. *

* This method will not 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 @@ -166,24 +145,23 @@ default void delete(Table table, TrackingRowSet rowSet) throws IOException { * cycle. * * @param table Table containing the rows to delete - * @param rowSet The rows to delete * @throws UnsupportedOperationException If this table does not support deletes */ - default void deleteAsync(Table table, TrackingRowSet rowSet, InputTableStatusListener listener) { + default void deleteAsync(Table table, InputTableStatusListener listener) { throw new UnsupportedOperationException("Table does not support deletes"); } /** - * Return a user-readable description of this MutableInputTable. + * 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 MutableInputTable. + * Returns a Deephaven table that contains the current data for this InputTable. * - * @return the current data in this MutableInputTable. + * @return the current data in this InputTable. */ Table getTable(); @@ -198,20 +176,20 @@ default boolean isKey(String columnName) { } /** - * Returns true if the specified column exists in this MutableInputTable. + * Returns true if the specified column exists in this InputTable. * * @param columnName the column to interrogate - * @return true if columnName exists in this MutableInputTable + * @return true if columnName exists in this InputTable */ default boolean hasColumn(String columnName) { return getTableDefinition().getColumnNames().contains(columnName); } /** - * Queries whether this MutableInputTable is editable in the current context. + * Queries whether this InputTable is editable in the current context. * - * @return true if this MutableInputTable may be edited, false otherwise TODO (deephaven/deephaven-core/issues/255): - * Add AuthContext and whatever else is appropriate + * @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(); } diff --git a/engine/table/src/test/java/io/deephaven/engine/table/impl/util/TestFunctionGeneratedTableFactory.java b/engine/table/src/test/java/io/deephaven/engine/table/impl/util/TestFunctionGeneratedTableFactory.java index d8cadb6b33f..b44c1d8d864 100644 --- a/engine/table/src/test/java/io/deephaven/engine/table/impl/util/TestFunctionGeneratedTableFactory.java +++ b/engine/table/src/test/java/io/deephaven/engine/table/impl/util/TestFunctionGeneratedTableFactory.java @@ -20,7 +20,7 @@ import java.util.Random; -import static io.deephaven.engine.table.impl.util.TestKeyedArrayBackedMutableTable.handleDelayedRefresh; +import static io.deephaven.engine.table.impl.util.TestKeyedArrayBackedInputTable.handleDelayedRefresh; import static io.deephaven.engine.testutil.TstUtils.*; import static io.deephaven.engine.util.TableTools.*; @@ -68,13 +68,13 @@ public void testNoSources() { } public void testMultipleSources() throws Exception { - final AppendOnlyArrayBackedMutableTable source1 = AppendOnlyArrayBackedMutableTable.make(TableDefinition.of( + final AppendOnlyArrayBackedInputTable source1 = AppendOnlyArrayBackedInputTable.make(TableDefinition.of( ColumnDefinition.of("StringCol", Type.stringType()))); - final BaseArrayBackedMutableTable.ArrayBackedMutableInputTable inputTable1 = source1.makeHandler(); + final BaseArrayBackedInputTable.ArrayBackedInputTableUpdater inputTable1 = source1.makeUpdater(); - final AppendOnlyArrayBackedMutableTable source2 = AppendOnlyArrayBackedMutableTable.make(TableDefinition.of( + final AppendOnlyArrayBackedInputTable source2 = AppendOnlyArrayBackedInputTable.make(TableDefinition.of( ColumnDefinition.of("IntCol", Type.intType()))); - final BaseArrayBackedMutableTable.ArrayBackedMutableInputTable inputTable2 = source2.makeHandler(); + final BaseArrayBackedInputTable.ArrayBackedInputTableUpdater inputTable2 = source2.makeUpdater(); final Table functionBacked = FunctionGeneratedTableFactory.create(() -> source1.lastBy().naturalJoin(source2, ""), source1, source2); @@ -82,9 +82,9 @@ public void testMultipleSources() throws Exception { assertEquals(functionBacked.size(), 0); handleDelayedRefresh(() -> { - inputTable1.addAsync(newTable(stringCol("StringCol", "MyString")), false, t -> { + inputTable1.addAsync(newTable(stringCol("StringCol", "MyString")), t -> { }); - inputTable2.addAsync(newTable(intCol("IntCol", 12345)), false, t -> { + inputTable2.addAsync(newTable(intCol("IntCol", 12345)), t -> { }); }, source1, source2); diff --git a/engine/table/src/test/java/io/deephaven/engine/table/impl/util/TestKeyedArrayBackedInputTable.java b/engine/table/src/test/java/io/deephaven/engine/table/impl/util/TestKeyedArrayBackedInputTable.java new file mode 100644 index 00000000000..72d468c4e8b --- /dev/null +++ b/engine/table/src/test/java/io/deephaven/engine/table/impl/util/TestKeyedArrayBackedInputTable.java @@ -0,0 +1,202 @@ +/** + * Copyright (c) 2016-2022 Deephaven Data Labs and Patent Pending + */ +package io.deephaven.engine.table.impl.util; + +import io.deephaven.UncheckedDeephavenException; +import io.deephaven.engine.context.ExecutionContext; +import io.deephaven.engine.table.Table; +import io.deephaven.engine.table.impl.FailureListener; +import io.deephaven.engine.table.impl.TableUpdateValidator; +import io.deephaven.engine.testutil.ControlledUpdateGraph; +import io.deephaven.engine.testutil.junit4.EngineCleanup; +import io.deephaven.engine.util.TableTools; +import io.deephaven.engine.util.input.InputTableUpdater; +import io.deephaven.util.function.ThrowingRunnable; +import junit.framework.TestCase; +import org.junit.Rule; +import org.junit.Test; + +import java.io.IOException; +import java.util.Arrays; +import java.util.concurrent.CountDownLatch; + +import static io.deephaven.engine.testutil.TstUtils.assertTableEquals; +import static io.deephaven.engine.util.TableTools.showWithRowSet; +import static io.deephaven.engine.util.TableTools.stringCol; + +public class TestKeyedArrayBackedInputTable { + + @Rule + public final EngineCleanup liveTableTestCase = new EngineCleanup(); + + @Test + public void testSimple() throws Exception { + final Table input = TableTools.newTable(stringCol("Name", "Fred", "George", "Earl"), + stringCol("Employer", "Slate Rock and Gravel", "Spacely Sprockets", "Wesayso")); + + final KeyedArrayBackedInputTable kabut = KeyedArrayBackedInputTable.make(input, "Name"); + final TableUpdateValidator validator = TableUpdateValidator.make("kabut", kabut); + final Table validatorResult = validator.getResultTable(); + final FailureListener failureListener = new FailureListener(); + validatorResult.addUpdateListener(failureListener); + + assertTableEquals(input, kabut); + + final InputTableUpdater inputTableUpdater = (InputTableUpdater) kabut.getAttribute(Table.INPUT_TABLE_ATTRIBUTE); + TestCase.assertNotNull(inputTableUpdater); + + final Table input2 = TableTools.newTable(stringCol("Name", "Randy"), stringCol("Employer", "USGS")); + + handleDelayedRefresh(() -> inputTableUpdater.add(input2), kabut); + assertTableEquals(TableTools.merge(input, input2), kabut); + + final Table input3 = TableTools.newTable(stringCol("Name", "Randy"), stringCol("Employer", "Tegridy")); + handleDelayedRefresh(() -> inputTableUpdater.add(input3), kabut); + assertTableEquals(TableTools.merge(input, input3), kabut); + + + final Table input4 = TableTools.newTable(stringCol("Name", "George"), stringCol("Employer", "Cogswell")); + handleDelayedRefresh(() -> inputTableUpdater.add(input4), kabut); + showWithRowSet(kabut); + + assertTableEquals(TableTools.merge(input, input3, input4).lastBy("Name"), kabut); + + final Table input5 = + TableTools.newTable(stringCol("Name", "George"), stringCol("Employer", "Spacely Sprockets")); + handleDelayedRefresh(() -> inputTableUpdater.add(input5), kabut); + showWithRowSet(kabut); + + assertTableEquals(TableTools.merge(input, input3, input4, input5).lastBy("Name"), kabut); + + final long sizeBeforeDelete = kabut.size(); + System.out.println("KABUT.rowSet before delete: " + kabut.getRowSet()); + final Table delete1 = TableTools.newTable(stringCol("Name", "Earl")); + handleDelayedRefresh(() -> inputTableUpdater.delete(delete1), kabut); + System.out.println("KABUT.rowSet after delete: " + kabut.getRowSet()); + final long sizeAfterDelete = kabut.size(); + TestCase.assertEquals(sizeBeforeDelete - 1, sizeAfterDelete); + + showWithRowSet(kabut); + + final Table expected = TableTools.merge( + TableTools.merge(input, input3, input4, input5).update("Deleted=false"), + delete1.update("Employer=(String)null", "Deleted=true")) + .lastBy("Name").where("Deleted=false").dropColumns("Deleted"); + showWithRowSet(expected); + + assertTableEquals(expected, kabut); + } + + @Test + public void testAppendOnly() throws Exception { + final Table input = TableTools.newTable(stringCol("Name", "Fred", "George", "Earl"), + stringCol("Employer", "Slate Rock and Gravel", "Spacely Sprockets", "Wesayso")); + + final AppendOnlyArrayBackedInputTable aoabmt = AppendOnlyArrayBackedInputTable.make(input); + final TableUpdateValidator validator = TableUpdateValidator.make("aoabmt", aoabmt); + final Table validatorResult = validator.getResultTable(); + final FailureListener failureListener = new FailureListener(); + validatorResult.addUpdateListener(failureListener); + + assertTableEquals(input, aoabmt); + + final InputTableUpdater inputTableUpdater = + (InputTableUpdater) aoabmt.getAttribute(Table.INPUT_TABLE_ATTRIBUTE); + TestCase.assertNotNull(inputTableUpdater); + + final Table input2 = + TableTools.newTable(stringCol("Name", "Randy", "George"), stringCol("Employer", "USGS", "Cogswell")); + + handleDelayedRefresh(() -> inputTableUpdater.add(input2), aoabmt); + assertTableEquals(TableTools.merge(input, input2), aoabmt); + } + + @Test + public void testFilteredAndSorted() throws Exception { + final Table input = TableTools.newTable(stringCol("Name", "Fred", "George", "Earl"), + stringCol("Employer", "Slate Rock and Gravel", "Spacely Sprockets", "Wesayso")); + + final KeyedArrayBackedInputTable kabut = KeyedArrayBackedInputTable.make(input, "Name"); + final TableUpdateValidator validator = TableUpdateValidator.make("kabut", kabut); + final Table validatorResult = validator.getResultTable(); + final FailureListener failureListener = new FailureListener(); + validatorResult.addUpdateListener(failureListener); + + assertTableEquals(input, kabut); + + final Table fs = kabut.where("Name.length() == 4").sort("Name"); + + final InputTableUpdater inputTableUpdater = (InputTableUpdater) fs.getAttribute(Table.INPUT_TABLE_ATTRIBUTE); + TestCase.assertNotNull(inputTableUpdater); + + final Table delete = TableTools.newTable(stringCol("Name", "Fred")); + + handleDelayedRefresh(() -> inputTableUpdater.delete(delete), kabut); + assertTableEquals(input.where("Name != `Fred`"), kabut); + } + + + @Test + public void testAddBack() throws Exception { + final Table input = TableTools.newTable(stringCol("Name"), stringCol("Employer")); + + final KeyedArrayBackedInputTable kabut = KeyedArrayBackedInputTable.make(input, "Name"); + final TableUpdateValidator validator = TableUpdateValidator.make("kabut", kabut); + final Table validatorResult = validator.getResultTable(); + final FailureListener failureListener = new FailureListener(); + validatorResult.addUpdateListener(failureListener); + + assertTableEquals(input, kabut); + + final InputTableUpdater inputTableUpdater = (InputTableUpdater) kabut.getAttribute(Table.INPUT_TABLE_ATTRIBUTE); + TestCase.assertNotNull(inputTableUpdater); + + final Table input2 = + TableTools.newTable(stringCol("Name", "George"), stringCol("Employer", "Spacely Sprockets")); + + handleDelayedRefresh(() -> inputTableUpdater.add(input2), kabut); + assertTableEquals(input2, kabut); + + handleDelayedRefresh(() -> inputTableUpdater.delete(input2.view("Name")), kabut); + assertTableEquals(input, kabut); + + handleDelayedRefresh(() -> inputTableUpdater.add(input2), kabut); + assertTableEquals(input2, kabut); + } + + public static void handleDelayedRefresh(final ThrowingRunnable action, + final BaseArrayBackedInputTable... tables) throws Exception { + final Thread refreshThread; + final CountDownLatch gate = new CountDownLatch(tables.length); + + Arrays.stream(tables).forEach(t -> t.setOnPendingChange(gate::countDown)); + try { + final ControlledUpdateGraph updateGraph = ExecutionContext.getContext().getUpdateGraph().cast(); + refreshThread = new Thread(() -> { + // If this unexpected interruption happens, the test thread may hang in action.run() + // indefinitely. Best to hope it's already queued the pending action and proceed with run. + updateGraph.runWithinUnitTestCycle(() -> { + try { + gate.await(); + } catch (InterruptedException ignored) { + // If this unexpected interruption happens, the test thread may hang in action.run() + // indefinitely. Best to hope it's already queued the pending action and proceed with run. + } + Arrays.stream(tables).forEach(BaseArrayBackedInputTable::run); + }); + }); + + refreshThread.start(); + action.run(); + } finally { + Arrays.stream(tables).forEach(t -> t.setOnPendingChange(null)); + } + try { + refreshThread.join(); + } catch (InterruptedException e) { + throw new UncheckedDeephavenException( + "Interrupted unexpectedly while waiting for run cycle to complete", e); + } + } +} diff --git a/engine/table/src/test/java/io/deephaven/engine/table/impl/util/TestKeyedArrayBackedMutableTable.java b/engine/table/src/test/java/io/deephaven/engine/table/impl/util/TestKeyedArrayBackedMutableTable.java deleted file mode 100644 index a211071cbe5..00000000000 --- a/engine/table/src/test/java/io/deephaven/engine/table/impl/util/TestKeyedArrayBackedMutableTable.java +++ /dev/null @@ -1,333 +0,0 @@ -/** - * Copyright (c) 2016-2022 Deephaven Data Labs and Patent Pending - */ -package io.deephaven.engine.table.impl.util; - -import io.deephaven.UncheckedDeephavenException; -import io.deephaven.base.SleepUtil; -import io.deephaven.datastructures.util.CollectionUtil; -import io.deephaven.engine.context.ExecutionContext; -import io.deephaven.engine.table.Table; -import io.deephaven.engine.table.impl.FailureListener; -import io.deephaven.engine.table.impl.TableUpdateValidator; -import io.deephaven.engine.testutil.ControlledUpdateGraph; -import io.deephaven.engine.testutil.junit4.EngineCleanup; -import io.deephaven.engine.util.TableTools; -import io.deephaven.engine.util.config.InputTableStatusListener; -import io.deephaven.engine.util.config.MutableInputTable; -import io.deephaven.util.function.ThrowingRunnable; -import junit.framework.TestCase; -import org.jetbrains.annotations.NotNull; -import org.jetbrains.annotations.Nullable; -import org.junit.Rule; -import org.junit.Test; - -import java.io.IOException; -import java.util.Arrays; -import java.util.Map; -import java.util.concurrent.CountDownLatch; - -import static io.deephaven.engine.testutil.TstUtils.assertTableEquals; -import static io.deephaven.engine.util.TableTools.showWithRowSet; -import static io.deephaven.engine.util.TableTools.stringCol; - -public class TestKeyedArrayBackedMutableTable { - - @Rule - public final EngineCleanup liveTableTestCase = new EngineCleanup(); - - @Test - public void testSimple() throws Exception { - final Table input = TableTools.newTable(stringCol("Name", "Fred", "George", "Earl"), - stringCol("Employer", "Slate Rock and Gravel", "Spacely Sprockets", "Wesayso")); - - final KeyedArrayBackedMutableTable kabut = KeyedArrayBackedMutableTable.make(input, "Name"); - final TableUpdateValidator validator = TableUpdateValidator.make("kabut", kabut); - final Table validatorResult = validator.getResultTable(); - final FailureListener failureListener = new FailureListener(); - validatorResult.addUpdateListener(failureListener); - - assertTableEquals(input, kabut); - - final MutableInputTable mutableInputTable = (MutableInputTable) kabut.getAttribute(Table.INPUT_TABLE_ATTRIBUTE); - TestCase.assertNotNull(mutableInputTable); - - final Table input2 = TableTools.newTable(stringCol("Name", "Randy"), stringCol("Employer", "USGS")); - - handleDelayedRefresh(() -> mutableInputTable.add(input2), kabut); - assertTableEquals(TableTools.merge(input, input2), kabut); - - final Table input3 = TableTools.newTable(stringCol("Name", "Randy"), stringCol("Employer", "Tegridy")); - handleDelayedRefresh(() -> mutableInputTable.add(input3), kabut); - assertTableEquals(TableTools.merge(input, input3), kabut); - - - final Table input4 = TableTools.newTable(stringCol("Name", "George"), stringCol("Employer", "Cogswell")); - handleDelayedRefresh(() -> mutableInputTable.add(input4), kabut); - showWithRowSet(kabut); - - assertTableEquals(TableTools.merge(input, input3, input4).lastBy("Name"), kabut); - - final Table input5 = - TableTools.newTable(stringCol("Name", "George"), stringCol("Employer", "Spacely Sprockets")); - handleDelayedRefresh(() -> mutableInputTable.add(input5), kabut); - showWithRowSet(kabut); - - assertTableEquals(TableTools.merge(input, input3, input4, input5).lastBy("Name"), kabut); - - final long sizeBeforeDelete = kabut.size(); - System.out.println("KABUT.rowSet before delete: " + kabut.getRowSet()); - final Table delete1 = TableTools.newTable(stringCol("Name", "Earl")); - handleDelayedRefresh(() -> mutableInputTable.delete(delete1), kabut); - System.out.println("KABUT.rowSet after delete: " + kabut.getRowSet()); - final long sizeAfterDelete = kabut.size(); - TestCase.assertEquals(sizeBeforeDelete - 1, sizeAfterDelete); - - showWithRowSet(kabut); - - final Table expected = TableTools.merge( - TableTools.merge(input, input3, input4, input5).update("Deleted=false"), - delete1.update("Employer=(String)null", "Deleted=true")) - .lastBy("Name").where("Deleted=false").dropColumns("Deleted"); - showWithRowSet(expected); - - assertTableEquals(expected, kabut); - } - - @Test - public void testAppendOnly() throws Exception { - final Table input = TableTools.newTable(stringCol("Name", "Fred", "George", "Earl"), - stringCol("Employer", "Slate Rock and Gravel", "Spacely Sprockets", "Wesayso")); - - final AppendOnlyArrayBackedMutableTable aoabmt = AppendOnlyArrayBackedMutableTable.make(input); - final TableUpdateValidator validator = TableUpdateValidator.make("aoabmt", aoabmt); - final Table validatorResult = validator.getResultTable(); - final FailureListener failureListener = new FailureListener(); - validatorResult.addUpdateListener(failureListener); - - assertTableEquals(input, aoabmt); - - final MutableInputTable mutableInputTable = - (MutableInputTable) aoabmt.getAttribute(Table.INPUT_TABLE_ATTRIBUTE); - TestCase.assertNotNull(mutableInputTable); - - final Table input2 = - TableTools.newTable(stringCol("Name", "Randy", "George"), stringCol("Employer", "USGS", "Cogswell")); - - handleDelayedRefresh(() -> mutableInputTable.add(input2), aoabmt); - assertTableEquals(TableTools.merge(input, input2), aoabmt); - } - - @Test - public void testFilteredAndSorted() throws Exception { - final Table input = TableTools.newTable(stringCol("Name", "Fred", "George", "Earl"), - stringCol("Employer", "Slate Rock and Gravel", "Spacely Sprockets", "Wesayso")); - - final KeyedArrayBackedMutableTable kabut = KeyedArrayBackedMutableTable.make(input, "Name"); - final TableUpdateValidator validator = TableUpdateValidator.make("kabut", kabut); - final Table validatorResult = validator.getResultTable(); - final FailureListener failureListener = new FailureListener(); - validatorResult.addUpdateListener(failureListener); - - assertTableEquals(input, kabut); - - final Table fs = kabut.where("Name.length() == 4").sort("Name"); - - final MutableInputTable mutableInputTable = (MutableInputTable) fs.getAttribute(Table.INPUT_TABLE_ATTRIBUTE); - TestCase.assertNotNull(mutableInputTable); - - final Table delete = TableTools.newTable(stringCol("Name", "Fred")); - - handleDelayedRefresh(() -> mutableInputTable.delete(delete), kabut); - assertTableEquals(input.where("Name != `Fred`"), kabut); - } - - @Test - public void testAddRows() throws Throwable { - final Table input = TableTools.newTable(stringCol("Name", "Fred", "George", "Earl"), - stringCol("Employer", "Slate Rock and Gravel", "Spacely Sprockets", "Wesayso")); - - final KeyedArrayBackedMutableTable kabut = KeyedArrayBackedMutableTable.make(input, "Name"); - final TableUpdateValidator validator = TableUpdateValidator.make("kabut", kabut); - final Table validatorResult = validator.getResultTable(); - final FailureListener failureListener = new FailureListener(); - validatorResult.addUpdateListener(failureListener); - - assertTableEquals(input, kabut); - - final MutableInputTable mutableInputTable = (MutableInputTable) kabut.getAttribute(Table.INPUT_TABLE_ATTRIBUTE); - TestCase.assertNotNull(mutableInputTable); - - final Table input2 = TableTools.newTable(stringCol("Name", "Randy"), stringCol("Employer", "USGS")); - - final Map randyMap = - CollectionUtil.mapFromArray(String.class, Object.class, "Name", "Randy", "Employer", "USGS"); - final TestStatusListener listener = new TestStatusListener(); - mutableInputTable.addRow(randyMap, true, listener); - SleepUtil.sleep(100); - listener.assertIncomplete(); - final ControlledUpdateGraph updateGraph = ExecutionContext.getContext().getUpdateGraph().cast(); - updateGraph.runWithinUnitTestCycle(kabut::run); - assertTableEquals(TableTools.merge(input, input2), kabut); - listener.waitForCompletion(); - listener.assertSuccess(); - - // TODO: should we throw the exception from the initial palce, should we defer edit checking to the UGP which - // would make it consistent, but also slower to produce errors and uglier for reporting? - final TestStatusListener listener2 = new TestStatusListener(); - final Map randyMap2 = - CollectionUtil.mapFromArray(String.class, Object.class, "Name", "Randy", "Employer", "Tegridy"); - mutableInputTable.addRow(randyMap2, false, listener2); - SleepUtil.sleep(100); - listener2.assertIncomplete(); - updateGraph.runWithinUnitTestCycle(kabut::run); - assertTableEquals(TableTools.merge(input, input2), kabut); - listener2.waitForCompletion(); - listener2.assertFailure(IllegalArgumentException.class, "Can not edit keys Randy"); - } - - @Test - public void testAddBack() throws Exception { - final Table input = TableTools.newTable(stringCol("Name"), stringCol("Employer")); - - final KeyedArrayBackedMutableTable kabut = KeyedArrayBackedMutableTable.make(input, "Name"); - final TableUpdateValidator validator = TableUpdateValidator.make("kabut", kabut); - final Table validatorResult = validator.getResultTable(); - final FailureListener failureListener = new FailureListener(); - validatorResult.addUpdateListener(failureListener); - - assertTableEquals(input, kabut); - - final MutableInputTable mutableInputTable = (MutableInputTable) kabut.getAttribute(Table.INPUT_TABLE_ATTRIBUTE); - TestCase.assertNotNull(mutableInputTable); - - final Table input2 = - TableTools.newTable(stringCol("Name", "George"), stringCol("Employer", "Spacely Sprockets")); - - handleDelayedRefresh(() -> mutableInputTable.add(input2), kabut); - assertTableEquals(input2, kabut); - - handleDelayedRefresh(() -> mutableInputTable.delete(input2.view("Name")), kabut); - assertTableEquals(input, kabut); - - handleDelayedRefresh(() -> mutableInputTable.add(input2), kabut); - assertTableEquals(input2, kabut); - } - - @Test - public void testSetRows() { - final Table input = TableTools.newTable(stringCol("Name", "Fred", "George", "Earl"), - stringCol("Employer", "Slate Rock and Gravel", "Spacely Sprockets", "Wesayso"), - stringCol("Spouse", "Wilma", "Jane", "Fran")); - - final KeyedArrayBackedMutableTable kabut = KeyedArrayBackedMutableTable.make(input, "Name"); - final TableUpdateValidator validator = TableUpdateValidator.make("kabut", kabut); - final Table validatorResult = validator.getResultTable(); - final FailureListener failureListener = new FailureListener(); - validatorResult.addUpdateListener(failureListener); - - assertTableEquals(input, kabut); - - final MutableInputTable mutableInputTable = (MutableInputTable) kabut.getAttribute(Table.INPUT_TABLE_ATTRIBUTE); - TestCase.assertNotNull(mutableInputTable); - - final Table defaultValues = input.where("Name=`George`"); - final Table ex2 = TableTools.newTable(stringCol("Name", "George"), stringCol("Employer", "Cogswell"), - stringCol("Spouse", "Jane")); - - final Map cogMap = - CollectionUtil.mapFromArray(String.class, Object.class, "Name", "George", "Employer", "Cogswell"); - mutableInputTable.setRow(defaultValues, 0, cogMap); - SleepUtil.sleep(100); - final ControlledUpdateGraph updateGraph = ExecutionContext.getContext().getUpdateGraph().cast(); - updateGraph.runWithinUnitTestCycle(kabut::run); - assertTableEquals(TableTools.merge(input, ex2).lastBy("Name"), kabut); - } - - private static class TestStatusListener implements InputTableStatusListener { - boolean success = false; - Throwable error = null; - - @Override - public synchronized void onError(Throwable t) { - if (success || error != null) { - throw new IllegalStateException("Can not complete listener twice!"); - } - error = t; - notifyAll(); - } - - @Override - public synchronized void onSuccess() { - if (success || error != null) { - throw new IllegalStateException("Can not complete listener twice!"); - } - success = true; - notifyAll(); - } - - private synchronized void assertIncomplete() { - TestCase.assertFalse(success); - TestCase.assertNull(error); - } - - private void waitForCompletion() throws InterruptedException { - synchronized (this) { - while (!success && error == null) { - wait(); - } - } - } - - private synchronized void assertSuccess() throws Throwable { - if (!success) { - throw error; - } - } - - private synchronized void assertFailure(@NotNull final Class errorClass, - @Nullable final String errorMessage) { - TestCase.assertFalse(success); - TestCase.assertNotNull(error); - TestCase.assertTrue(errorClass.isAssignableFrom(error.getClass())); - if (errorMessage != null) { - TestCase.assertEquals(errorMessage, error.getMessage()); - } - } - } - - public static void handleDelayedRefresh(final ThrowingRunnable action, - final BaseArrayBackedMutableTable... tables) throws Exception { - final Thread refreshThread; - final CountDownLatch gate = new CountDownLatch(tables.length); - - Arrays.stream(tables).forEach(t -> t.setOnPendingChange(gate::countDown)); - try { - final ControlledUpdateGraph updateGraph = ExecutionContext.getContext().getUpdateGraph().cast(); - refreshThread = new Thread(() -> { - // If this unexpected interruption happens, the test thread may hang in action.run() - // indefinitely. Best to hope it's already queued the pending action and proceed with run. - updateGraph.runWithinUnitTestCycle(() -> { - try { - gate.await(); - } catch (InterruptedException ignored) { - // If this unexpected interruption happens, the test thread may hang in action.run() - // indefinitely. Best to hope it's already queued the pending action and proceed with run. - } - Arrays.stream(tables).forEach(BaseArrayBackedMutableTable::run); - }); - }); - - refreshThread.start(); - action.run(); - } finally { - Arrays.stream(tables).forEach(t -> t.setOnPendingChange(null)); - } - try { - refreshThread.join(); - } catch (InterruptedException e) { - throw new UncheckedDeephavenException( - "Interrupted unexpectedly while waiting for run cycle to complete", e); - } - } -} diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/util/BarrageUtil.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/util/BarrageUtil.java index 298e97f988f..3b4a1c11423 100755 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/util/BarrageUtil.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/util/BarrageUtil.java @@ -36,7 +36,7 @@ import io.deephaven.proto.util.Exceptions; import io.deephaven.api.util.NameValidator; import io.deephaven.engine.util.ColumnFormatting; -import io.deephaven.engine.util.config.MutableInputTable; +import io.deephaven.engine.util.input.InputTableUpdater; import io.deephaven.chunk.ChunkType; import io.deephaven.proto.backplane.grpc.ExportedTableCreationResponse; import io.deephaven.util.type.TypeUtils; @@ -148,9 +148,10 @@ public static int makeTableSchemaPayload( final Map schemaMetadata = attributesToMetadata(attributes); final Map descriptions = GridAttributes.getColumnDescriptions(attributes); - final MutableInputTable inputTable = (MutableInputTable) attributes.get(Table.INPUT_TABLE_ATTRIBUTE); + final InputTableUpdater inputTableUpdater = (InputTableUpdater) attributes.get(Table.INPUT_TABLE_ATTRIBUTE); final List fields = columnDefinitionsToFields( - descriptions, inputTable, tableDefinition, tableDefinition.getColumns(), ignored -> new HashMap<>(), + descriptions, inputTableUpdater, tableDefinition, tableDefinition.getColumns(), + ignored -> new HashMap<>(), attributes, options.columnsAsList()) .collect(Collectors.toList()); @@ -180,12 +181,12 @@ public static Map attributesToMetadata(@NotNull final Map columnDefinitionsToFields( @NotNull final Map columnDescriptions, - @Nullable final MutableInputTable inputTable, + @Nullable final InputTableUpdater inputTableUpdater, @NotNull final TableDefinition tableDefinition, @NotNull final Collection> columnDefinitions, @NotNull final Function> fieldMetadataFactory, @NotNull final Map attributes) { - return columnDefinitionsToFields(columnDescriptions, inputTable, tableDefinition, columnDefinitions, + return columnDefinitionsToFields(columnDescriptions, inputTableUpdater, tableDefinition, columnDefinitions, fieldMetadataFactory, attributes, false); @@ -197,7 +198,7 @@ private static boolean isDataTypeSortable(final Class dataType) { public static Stream columnDefinitionsToFields( @NotNull final Map columnDescriptions, - @Nullable final MutableInputTable inputTable, + @Nullable final InputTableUpdater inputTableUpdater, @NotNull final TableDefinition tableDefinition, @NotNull final Collection> columnDefinitions, @NotNull final Function> fieldMetadataFactory, @@ -274,8 +275,8 @@ public static Stream columnDefinitionsToFields( if (columnDescription != null) { putMetadata(metadata, "description", columnDescription); } - if (inputTable != null) { - putMetadata(metadata, "inputtable.isKey", inputTable.getKeyNames().contains(name) + ""); + if (inputTableUpdater != null) { + putMetadata(metadata, "inputtable.isKey", inputTableUpdater.getKeyNames().contains(name) + ""); } if (columnsAsList) { diff --git a/py/server/deephaven/table_factory.py b/py/server/deephaven/table_factory.py index 033d4c7aec2..5dc5e934f17 100644 --- a/py/server/deephaven/table_factory.py +++ b/py/server/deephaven/table_factory.py @@ -24,10 +24,9 @@ _JTableFactory = jpy.get_type("io.deephaven.engine.table.TableFactory") _JTableTools = jpy.get_type("io.deephaven.engine.util.TableTools") _JDynamicTableWriter = jpy.get_type("io.deephaven.engine.table.impl.util.DynamicTableWriter") -_JMutableInputTable = jpy.get_type("io.deephaven.engine.util.config.MutableInputTable") -_JAppendOnlyArrayBackedMutableTable = jpy.get_type( - "io.deephaven.engine.table.impl.util.AppendOnlyArrayBackedMutableTable") -_JKeyedArrayBackedMutableTable = jpy.get_type("io.deephaven.engine.table.impl.util.KeyedArrayBackedMutableTable") +_JAppendOnlyArrayBackedInputTable = jpy.get_type( + "io.deephaven.engine.table.impl.util.AppendOnlyArrayBackedInputTable") +_JKeyedArrayBackedInputTable = jpy.get_type("io.deephaven.engine.table.impl.util.KeyedArrayBackedInputTable") _JTableDefinition = jpy.get_type("io.deephaven.engine.table.TableDefinition") _JTable = jpy.get_type("io.deephaven.engine.table.Table") _J_INPUT_TABLE_ATTRIBUTE = _JTable.INPUT_TABLE_ATTRIBUTE @@ -257,9 +256,9 @@ def __init__(self, col_defs: Dict[str, DType] = None, init_table: Table = None, key_cols = to_sequence(key_cols) if key_cols: - super().__init__(_JKeyedArrayBackedMutableTable.make(j_arg_1, key_cols)) + super().__init__(_JKeyedArrayBackedInputTable.make(j_arg_1, key_cols)) else: - super().__init__(_JAppendOnlyArrayBackedMutableTable.make(j_arg_1)) + super().__init__(_JAppendOnlyArrayBackedInputTable.make(j_arg_1)) self.j_input_table = self.j_table.getAttribute(_J_INPUT_TABLE_ATTRIBUTE) self.key_columns = key_cols except Exception as e: diff --git a/server/src/main/java/io/deephaven/server/table/inputtables/InputTableServiceGrpcImpl.java b/server/src/main/java/io/deephaven/server/table/inputtables/InputTableServiceGrpcImpl.java index 673e39e35b1..1436d7b6af4 100644 --- a/server/src/main/java/io/deephaven/server/table/inputtables/InputTableServiceGrpcImpl.java +++ b/server/src/main/java/io/deephaven/server/table/inputtables/InputTableServiceGrpcImpl.java @@ -10,7 +10,7 @@ import io.deephaven.engine.table.TableDefinition; import io.deephaven.engine.table.impl.perf.QueryPerformanceNugget; import io.deephaven.engine.table.impl.perf.QueryPerformanceRecorder; -import io.deephaven.engine.util.config.MutableInputTable; +import io.deephaven.engine.util.input.InputTableUpdater; import io.deephaven.extensions.barrage.util.GrpcUtil; import io.deephaven.internal.log.LoggerFactory; import io.deephaven.io.logger.Logger; @@ -74,13 +74,13 @@ public void addTableToInputTable( .onError(responseObserver) .require(targetTable, tableToAddExport) .submit(() -> { - Object inputTable = targetTable.get().getAttribute(Table.INPUT_TABLE_ATTRIBUTE); - if (!(inputTable instanceof MutableInputTable)) { + Object inputTableAsObject = targetTable.get().getAttribute(Table.INPUT_TABLE_ATTRIBUTE); + if (!(inputTableAsObject instanceof InputTableUpdater)) { throw Exceptions.statusRuntimeException(Code.INVALID_ARGUMENT, "Table can't be used as an input table"); } - MutableInputTable mutableInputTable = (MutableInputTable) inputTable; + final InputTableUpdater inputTableUpdater = (InputTableUpdater) inputTableAsObject; Table tableToAdd = tableToAddExport.get(); authWiring.checkPermissionAddTableToInputTable( @@ -89,7 +89,7 @@ public void addTableToInputTable( // validate that the columns are compatible try { - mutableInputTable.validateAddOrModify(tableToAdd); + inputTableUpdater.validateAddOrModify(tableToAdd); } catch (TableDefinition.IncompatibleTableDefinitionException exception) { throw Exceptions.statusRuntimeException(Code.INVALID_ARGUMENT, "Provided tables's columns are not compatible: " + exception.getMessage()); @@ -97,7 +97,7 @@ public void addTableToInputTable( // actually add the tables contents try { - mutableInputTable.add(tableToAdd); + inputTableUpdater.add(tableToAdd); GrpcUtil.safelyComplete(responseObserver, AddTableResponse.getDefaultInstance()); } catch (IOException ioException) { throw Exceptions.statusRuntimeException(Code.DATA_LOSS, @@ -132,13 +132,13 @@ public void deleteTableFromInputTable( .onError(responseObserver) .require(targetTable, tableToRemoveExport) .submit(() -> { - Object inputTable = targetTable.get().getAttribute(Table.INPUT_TABLE_ATTRIBUTE); - if (!(inputTable instanceof MutableInputTable)) { + Object inputTableAsObject = targetTable.get().getAttribute(Table.INPUT_TABLE_ATTRIBUTE); + if (!(inputTableAsObject instanceof InputTableUpdater)) { throw Exceptions.statusRuntimeException(Code.INVALID_ARGUMENT, "Table can't be used as an input table"); } - MutableInputTable mutableInputTable = (MutableInputTable) inputTable; + final InputTableUpdater inputTableUpdater = (InputTableUpdater) inputTableAsObject; Table tableToRemove = tableToRemoveExport.get(); authWiring.checkPermissionDeleteTableFromInputTable( @@ -147,7 +147,7 @@ public void deleteTableFromInputTable( // validate that the columns are compatible try { - mutableInputTable.validateDelete(tableToRemove); + inputTableUpdater.validateDelete(tableToRemove); } catch (TableDefinition.IncompatibleTableDefinitionException exception) { throw Exceptions.statusRuntimeException(Code.INVALID_ARGUMENT, "Provided tables's columns are not compatible: " + exception.getMessage()); @@ -158,7 +158,7 @@ public void deleteTableFromInputTable( // actually delete the table's contents try { - mutableInputTable.delete(tableToRemove); + inputTableUpdater.delete(tableToRemove); GrpcUtil.safelyComplete(responseObserver, DeleteTableResponse.getDefaultInstance()); } catch (IOException ioException) { throw Exceptions.statusRuntimeException(Code.DATA_LOSS, diff --git a/server/src/main/java/io/deephaven/server/table/ops/CreateInputTableGrpcImpl.java b/server/src/main/java/io/deephaven/server/table/ops/CreateInputTableGrpcImpl.java index 717542465a8..981b7ea76a6 100644 --- a/server/src/main/java/io/deephaven/server/table/ops/CreateInputTableGrpcImpl.java +++ b/server/src/main/java/io/deephaven/server/table/ops/CreateInputTableGrpcImpl.java @@ -8,8 +8,8 @@ import io.deephaven.datastructures.util.CollectionUtil; import io.deephaven.engine.table.Table; import io.deephaven.engine.table.TableDefinition; -import io.deephaven.engine.table.impl.util.AppendOnlyArrayBackedMutableTable; -import io.deephaven.engine.table.impl.util.KeyedArrayBackedMutableTable; +import io.deephaven.engine.table.impl.util.AppendOnlyArrayBackedInputTable; +import io.deephaven.engine.table.impl.util.KeyedArrayBackedInputTable; import io.deephaven.extensions.barrage.util.BarrageUtil; import io.deephaven.proto.backplane.grpc.BatchTableRequest; import io.deephaven.proto.backplane.grpc.CreateInputTableRequest; @@ -71,9 +71,9 @@ public Table create(final CreateInputTableRequest request, switch (request.getKind().getKindCase()) { case IN_MEMORY_APPEND_ONLY: - return AppendOnlyArrayBackedMutableTable.make(tableDefinitionFromSchema); + return AppendOnlyArrayBackedInputTable.make(tableDefinitionFromSchema); case IN_MEMORY_KEY_BACKED: - return KeyedArrayBackedMutableTable.make(tableDefinitionFromSchema, + return KeyedArrayBackedInputTable.make(tableDefinitionFromSchema, request.getKind().getInMemoryKeyBacked().getKeyColumnsList() .toArray(CollectionUtil.ZERO_LENGTH_STRING_ARRAY)); case KIND_NOT_SET: