diff --git a/Base/build.gradle b/Base/build.gradle
index 87f454fd3bf..f2d2fae4bf1 100644
--- a/Base/build.gradle
+++ b/Base/build.gradle
@@ -1,6 +1,8 @@
 dependencies {
     compile depLog4j, depTrove3, depAnnotations, depCommonsCompress
 
+    compile depCommonsLang3
+
     compile 'io.deephaven:hash:0.1.0'
 
     testCompile fileTree(dir: "${rootDir}/test-libs", include: ['*.jar'])
diff --git a/Base/src/main/java/io/deephaven/base/ClassUtil.java b/Base/src/main/java/io/deephaven/base/ClassUtil.java
index 78bde6be23b..bfc3a1e3564 100644
--- a/Base/src/main/java/io/deephaven/base/ClassUtil.java
+++ b/Base/src/main/java/io/deephaven/base/ClassUtil.java
@@ -4,78 +4,38 @@
 
 package io.deephaven.base;
 
-import org.apache.log4j.Logger;
+import org.apache.commons.lang3.ClassUtils;
 
-import java.lang.reflect.Array;
-import java.lang.reflect.Field;
-import java.lang.reflect.Modifier;
-import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 public final class ClassUtil {
-    public static String getBaseName(final String s) {
-        int i = s.lastIndexOf(".");
-        return (i == -1) ? s : s.substring(i + 1);
-    }
-
-    public static void dumpFinals(final Logger log, final String prefix, final Object p) {
-        final Class c = p.getClass();
-        Field[] fields = c.getDeclaredFields();
-        final int desiredMods = (Modifier.PUBLIC | Modifier.FINAL);
-        for (Field f : fields) {
-            if ((f.getModifiers() & desiredMods) == 0)
-                continue;
-            try {
-                final String tName = f.getType().getName();
-                final String name = f.getName();
-                final Object value = f.get(p);
-                log.info(prefix
-                        + tName
-                        + " " + name
-                        + " = " + value.toString());
-            } catch (Exception ignored) {
-            }
-        }
-    }
 
     public static <T> Class<T> generify(Class c) {
         return c;
     }
 
-    private static final Map<String, Class<?>> classMap = new HashMap<>();
-    public static Map<String, Class<?>> primitives = new HashMap<>();
-
-    static {
-        primitives.put("boolean", boolean.class);
-        primitives.put("int", int.class);
-        primitives.put("double", double.class);
-        primitives.put("long", long.class);
-        primitives.put("byte", byte.class);
-        primitives.put("short", short.class);
-        primitives.put("char", char.class);
-        primitives.put("float", float.class);
-    }
+    private static final Map<String, Class<?>> classMap = new ConcurrentHashMap<>();
 
     private static Class<?> getJavaType(String selectedType) throws ClassNotFoundException {
-        int arrayCount = 0;
-        while (selectedType.endsWith("[]")) {
-            selectedType = selectedType.substring(0, selectedType.length() - 2);
-            ++arrayCount;
-        }
-        Class<?> result = primitives.get(selectedType);
-        if (result == null && selectedType.startsWith("java.lang.")) {
-            result = primitives.get(selectedType.substring("java.lang.".length()));
-        }
-        if (result == null) {
-            result = Class.forName(selectedType.split("<")[0]);
-        }
-        if (arrayCount > 0) {
-            final int[] dimensions = new int[arrayCount];
-            result = Array.newInstance(result, dimensions).getClass();
-        }
-        return result;
+        // Given string might have generics, remove those before delegating to
+        // commons-lang3 for lookup implementation. Greedily match from first
+        // '<' to last '>' and remove all, so that the two types of array
+        // notation are retained.
+        String noGenerics = selectedType.replaceAll("<.*>", "");
+
+        return ClassUtils.getClass(noGenerics, false);
     }
 
+    /**
+     * Finds and caches Class instances based on name. This implementation can handle the strings created by
+     * {@link Class#getName()} and {@link Class#getCanonicalName()}, and some mixture of the two. JNI names are not
+     * supported.
+     *
+     * @param name the name of the class to lookup.
+     * @return A class instance
+     * @throws ClassNotFoundException if the class cannot be found
+     */
     public static Class<?> lookupClass(final String name) throws ClassNotFoundException {
         Class<?> result = classMap.get(name);
         if (result == null) {
@@ -83,12 +43,16 @@ public static Class<?> lookupClass(final String name) throws ClassNotFoundExcept
                 result = getJavaType(name);
                 classMap.put(name, result);
             } catch (ClassNotFoundException e) {
-                classMap.put(name, ClassUtil.class);
+                // Note that this prevents some runtime fix to the classpath and retrying
+                classMap.put(name, FailedToResolve.class);
                 throw e;
             }
-        } else if (result == ClassUtil.class) {
+        } else if (result == FailedToResolve.class) {
             throw new ClassNotFoundException(name);
         }
         return result;
     }
+
+    private static class FailedToResolve {
+    }
 }
diff --git a/Base/src/test/java/io/deephaven/base/ClassUtilTest.java b/Base/src/test/java/io/deephaven/base/ClassUtilTest.java
new file mode 100644
index 00000000000..3d0a2f3d598
--- /dev/null
+++ b/Base/src/test/java/io/deephaven/base/ClassUtilTest.java
@@ -0,0 +1,116 @@
+package io.deephaven.base;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+
+@RunWith(Parameterized.class)
+public class ClassUtilTest {
+    @Parameterized.Parameters
+    public static List<Function<Class<?>, String>[]> naming() {
+        return Arrays.asList(
+                new Function[] {(Function<Class<?>, String>) Class::getName},
+                new Function[] {(Function<Class<?>, String>) Class::getCanonicalName});
+    }
+
+    private final Function<Class<?>, String> namer;
+
+    public ClassUtilTest(Function<Class<?>, String> namer) {
+        this.namer = namer;
+    }
+
+    private Class<?> lookup(Class<?> clazz) throws ClassNotFoundException {
+        String string = namer.apply(clazz);
+        System.out.println(string);
+        return ClassUtil.lookupClass(string);
+    }
+
+    private void assertRoundTrip(Class<?> clazz) throws ClassNotFoundException {
+        assertSame(clazz, lookup(clazz));
+    }
+
+    @Test
+    public void lookupClassSimple() throws ClassNotFoundException {
+        assertRoundTrip(String.class);
+        assertRoundTrip(ClassUtilTest.class);
+    }
+
+    @Test
+    public void lookupClassPrimitive() throws ClassNotFoundException {
+        assertRoundTrip(int.class);
+        assertRoundTrip(Integer.class);
+        assertRoundTrip(char.class);
+
+        // assertRoundTrip(void.class);
+        assertRoundTrip(Void.class);
+    }
+
+    @Test
+    public void lookupClassArray() throws ClassNotFoundException {
+        assertRoundTrip(String[].class);
+        assertRoundTrip(String[][][][][][].class);
+        assertRoundTrip(ClassUtilTest[][][][][][].class);
+
+        assertRoundTrip(int[].class);
+    }
+
+    @Test
+    public void lookupClassInner() throws ClassNotFoundException {
+        assertRoundTrip(StaticInnerClass.class);
+        assertRoundTrip(InnerClass.class);
+        assertRoundTrip(StaticInnerClass.StaticInnerStaticInnerClass.class);
+        assertRoundTrip(StaticInnerClass.InnerStaticInnerClass.class);
+        assertRoundTrip(InnerClass.InnerInnerClass.class);
+        assertRoundTrip(Outer.class);
+    }
+
+    @Test
+    public void lookupClassWithClinit() throws ClassNotFoundException {
+        assertRoundTrip(VerifyNotInitialized.class);
+        assertFalse(verifyNotInitializedWasNotInitialized);
+    }
+
+    @Test
+    public void testGenericStrings() throws ClassNotFoundException {
+        assertSame(Map.Entry.class, ClassUtil.lookupClass(namer.apply(Map.Entry.class) + "<String, String>"));
+        // note that this name isn't quite legal for getName(), will try a few iterations to make sure we DWIM
+        assertSame(Map.Entry[].class, ClassUtil.lookupClass(namer.apply(Map.Entry.class) + "<String, String>[]"));
+        assertSame(Map.Entry[].class, ClassUtil.lookupClass(namer.apply(Map.Entry[].class) + "<String, String>"));
+        assertSame(Map.Entry[].class, ClassUtil.lookupClass("[L" + namer.apply(Map.Entry.class) + "<String, String>;"));
+    }
+
+    public static class StaticInnerClass {
+
+        public class StaticInnerStaticInnerClass {
+        }
+        public static class InnerStaticInnerClass {
+        }
+    }
+
+    public class InnerClass {
+        public class InnerInnerClass {
+        }
+    }
+
+    // Do not reset this by hand, that won't cause the class to be re-initialized
+    public static boolean verifyNotInitializedWasNotInitialized = false;
+}
+
+
+class Outer {
+}
+
+
+class VerifyNotInitialized {
+    static {
+        ClassUtilTest.verifyNotInitializedWasNotInitialized = true;
+    }
+}
diff --git a/DB/src/main/java/io/deephaven/db/tables/TableDefinition.java b/DB/src/main/java/io/deephaven/db/tables/TableDefinition.java
index c25162c665a..bd8db943038 100644
--- a/DB/src/main/java/io/deephaven/db/tables/TableDefinition.java
+++ b/DB/src/main/java/io/deephaven/db/tables/TableDefinition.java
@@ -234,6 +234,7 @@ public String getColumnNamesAsString() {
      *
      * @param other the other definition
      * @return {@code this} table definition, but in the the column order of {@code other}
+     * @throws IncompatibleTableDefinitionException if the definitions are not compatible
      */
     public TableDefinition checkMutualCompatibility(@NotNull final TableDefinition other) {
         TableDefinition result = checkCompatibility(other, false);
diff --git a/DB/src/main/java/io/deephaven/db/util/config/MutableInputTable.java b/DB/src/main/java/io/deephaven/db/util/config/MutableInputTable.java
index 456b23c773c..8af94440357 100644
--- a/DB/src/main/java/io/deephaven/db/util/config/MutableInputTable.java
+++ b/DB/src/main/java/io/deephaven/db/util/config/MutableInputTable.java
@@ -1,31 +1,41 @@
 package io.deephaven.db.util.config;
 
+import io.deephaven.db.exceptions.ArgumentException;
+import io.deephaven.db.tables.ColumnDefinition;
 import io.deephaven.db.tables.Table;
 import io.deephaven.db.tables.TableDefinition;
 import io.deephaven.db.v2.utils.Index;
-import io.deephaven.web.shared.data.InputTableDefinition;
 
 import java.io.IOException;
-import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
 
 /**
- * A minimal interface for mutable tables that can be changed over the OpenAPI.
+ * 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.
+ *
+ * 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 {
+
     /**
-     * Get the key and value columns names for this table.
+     * Gets the names of the key columns.
      *
-     * @return the InputTableDefinition.
+     * @return a list with the names of the key columns of this input table
      */
-    InputTableDefinition getDefinition();
+    List<String> getKeyNames();
 
     /**
-     * Get the names of the key columns
+     * Gets the names of the value columns. By default, any column not marked as a key column is a value column.
      *
-     * @return an array with the names of our key columns
+     * @return a list with the names of the value columns of this input table
      */
-    default String[] getKeyNames() {
-        return getDefinition().getKeys();
+    default List<String> getValueNames() {
+        List<String> keyNames = getKeyNames();
+        return getTableDefinition().getColumnNames().stream()
+                .filter(colName -> !keyNames.contains(colName))
+                .collect(Collectors.toList());
     }
 
     /**
@@ -36,7 +46,50 @@ default String[] getKeyNames() {
     TableDefinition getTableDefinition();
 
     /**
-     * Write newData to this table.
+     * Helper to check if a table is compatible with this table, so that it could be added as contents.
+     *
+     * @param tableToApply the table to check if it can used to add or modify this input table
+     * @throws TableDefinition.IncompatibleTableDefinitionException if the definitions are not compatible
+     */
+    default void validateAddOrModify(final Table tableToApply) {
+        getTableDefinition().checkMutualCompatibility(tableToApply.getDefinition());
+    }
+
+    /**
+     * Validates that the given table definition is suitable to be passed to {@link #delete(Table)}.
+     *
+     * @param tableToDelete the definition of the table to delete
+     * @throws UnsupportedOperationException if this table does not support deletes
+     * @throws io.deephaven.db.exceptions.ArgumentException if the given definition isn't compatible to be used to
+     *         delete
+     */
+    default void validateDelete(Table tableToDelete) {
+        final TableDefinition keyDefinition = tableToDelete.getDefinition();
+        final TableDefinition thisDefinition = getTableDefinition();
+        final StringBuilder error = new StringBuilder();
+        final List<String> keyNames = getKeyNames();
+        for (String keyColumn : keyNames) {
+            final ColumnDefinition<?> colDef = keyDefinition.getColumn(keyColumn);
+            final ColumnDefinition<?> thisColDef = thisDefinition.getColumn(keyColumn);
+            if (colDef == null) {
+                error.append("Key Column \"").append(keyColumn).append("\" does not exist.\n");
+            } else if (!colDef.isCompatible(thisColDef)) {
+                error.append("Key Column \"").append(keyColumn).append("\" is not compatible.\n");
+            }
+        }
+        final List<String> extraKeys = keyDefinition.getColumnNames().stream().filter(kd -> !keyNames.contains(kd))
+                .collect(Collectors.toList());
+        if (!extraKeys.isEmpty()) {
+            error.append("Unknown key columns: ").append(extraKeys);
+        }
+        if (error.length() > 0) {
+            throw new ArgumentException("Invalid Key Table Definition: " + error.toString());
+        }
+    }
+
+    /**
+     * Write newData to this table. This method will block until the rows are added. Added rows with keys that match
+     * existing rows will instead replace those rows, if supported.
      *
      * @param newData the data to write to this table
      *
@@ -50,6 +103,7 @@ default String[] getKeyNames() {
      *
      * @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.getIndex());
@@ -60,9 +114,12 @@ default void delete(Table table) throws IOException {
      * deleted.
      *
      * @param table The rows to delete.
-     * @throws IOException If a problem occurred while deleting the rows.
+     * @throws IOException if a problem occurred while deleting the rows
+     * @throws UnsupportedOperationException if this table does not support deletes
      */
-    void delete(Table table, Index index) throws IOException;
+    default void delete(Table table, Index index) throws IOException {
+        throw new UnsupportedOperationException("Table does not support deletes");
+    }
 
     /**
      * Return a user-readable description of this MutableInputTable.
@@ -86,7 +143,7 @@ default void delete(Table table) throws IOException {
      * @return true if columnName is a key column, false otherwise
      */
     default boolean isKey(String columnName) {
-        return Arrays.asList(getDefinition().getKeys()).contains(columnName);
+        return getKeyNames().contains(columnName);
     }
 
     /**
@@ -97,7 +154,7 @@ default boolean isKey(String columnName) {
      * @return true if columnName exists in this MutableInputTable
      */
     default boolean hasColumn(String columnName) {
-        return isKey(columnName) || Arrays.asList(getDefinition().getValues()).contains(columnName);
+        return getTableDefinition().getColumnNames().contains(columnName);
     }
 
     /**
diff --git a/DB/src/main/java/io/deephaven/db/v2/utils/AppendOnlyArrayBackedMutableTable.java b/DB/src/main/java/io/deephaven/db/v2/utils/AppendOnlyArrayBackedMutableTable.java
index 128c4997f86..0c1f20acc9e 100644
--- a/DB/src/main/java/io/deephaven/db/v2/utils/AppendOnlyArrayBackedMutableTable.java
+++ b/DB/src/main/java/io/deephaven/db/v2/utils/AppendOnlyArrayBackedMutableTable.java
@@ -14,6 +14,7 @@
 import org.jetbrains.annotations.NotNull;
 
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
 
@@ -82,8 +83,6 @@ public static AppendOnlyArrayBackedMutableTable make(final Table initialTable,
     private AppendOnlyArrayBackedMutableTable(@NotNull TableDefinition definition,
             final Map<String, Object[]> enumValues, final ProcessPendingUpdater processPendingUpdater) {
         super(Index.FACTORY.getEmptyIndex(), makeColumnSourceMap(definition), enumValues, processPendingUpdater);
-        inputTableDefinition.setKeys(CollectionUtil.ZERO_LENGTH_STRING_ARRAY);
-        inputTableDefinition.setValues(definition.getColumnNamesArray());
     }
 
     @Override
@@ -116,7 +115,7 @@ protected void processPendingTable(Table table, boolean allowEdits, IndexChangeR
 
     @Override
     protected void processPendingDelete(Table table, IndexChangeRecorder indexChangeRecorder) {
-        throw new UnsupportedOperationException();
+        throw new UnsupportedOperationException("Table doesn't support delete operation");
     }
 
     @Override
@@ -125,8 +124,8 @@ protected String getDefaultDescription() {
     }
 
     @Override
-    void validateDelete(final TableDefinition keyDefinition) {
-        throw new UnsupportedOperationException();
+    protected List<String> getKeyNames() {
+        return Collections.emptyList();
     }
 
     @Override
@@ -136,14 +135,14 @@ ArrayBackedMutableInputTable makeHandler() {
 
     private class AppendOnlyArrayBackedMutableInputTable extends ArrayBackedMutableInputTable {
         @Override
-        public void delete(Table table, Index index) {
+        public void setRows(@NotNull Table defaultValues, int[] rowArray, Map<String, Object>[] valueArray,
+                InputTableStatusListener listener) {
             throw new UnsupportedOperationException();
         }
 
         @Override
-        public void setRows(@NotNull Table defaultValues, int[] rowArray, Map<String, Object>[] valueArray,
-                InputTableStatusListener listener) {
-            throw new UnsupportedOperationException();
+        public void validateDelete(Table tableToDelete) {
+            throw new UnsupportedOperationException("Table doesn't support delete operation");
         }
 
         @Override
diff --git a/DB/src/main/java/io/deephaven/db/v2/utils/BaseArrayBackedMutableTable.java b/DB/src/main/java/io/deephaven/db/v2/utils/BaseArrayBackedMutableTable.java
index 62e1ff6e34f..95c5cc96de8 100644
--- a/DB/src/main/java/io/deephaven/db/v2/utils/BaseArrayBackedMutableTable.java
+++ b/DB/src/main/java/io/deephaven/db/v2/utils/BaseArrayBackedMutableTable.java
@@ -15,9 +15,9 @@
 import io.deephaven.db.v2.sources.ArrayBackedColumnSource;
 import io.deephaven.db.v2.sources.ColumnSource;
 import io.deephaven.util.annotations.TestUseOnly;
-import io.deephaven.web.shared.data.InputTableDefinition;
 import org.jetbrains.annotations.NotNull;
 
+import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicLong;
@@ -27,7 +27,6 @@ abstract class BaseArrayBackedMutableTable extends UpdatableTable {
 
     private static final Object[] BOOLEAN_ENUM_ARRAY = new Object[] {true, false, null};
 
-    protected final InputTableDefinition inputTableDefinition;
     private final List<PendingChange> pendingChanges = Collections.synchronizedList(new ArrayList<>());
     private final AtomicLong nextSequence = new AtomicLong(0);
     private final AtomicLong processedSequence = new AtomicLong(0);
@@ -43,7 +42,6 @@ public BaseArrayBackedMutableTable(Index index, Map<String, ? extends ColumnSour
             Map<String, Object[]> enumValues, ProcessPendingUpdater processPendingUpdater) {
         super(index, nameToColumnSource, processPendingUpdater);
         this.enumValues = enumValues;
-        this.inputTableDefinition = new InputTableDefinition();
         MutableInputTable mutableInputTable = makeHandler();
         setAttribute(Table.INPUT_TABLE_ATTRIBUTE, mutableInputTable);
         setRefreshing(true);
@@ -132,13 +130,7 @@ protected abstract void processPendingTable(Table table, boolean allowEdits,
 
     protected abstract String getDefaultDescription();
 
-    abstract void validateDelete(TableDefinition definition);
-
-    private void validateDefinition(final TableDefinition newDefinition) {
-        final TableDefinition thisDefinition = getDefinition();
-        thisDefinition.checkCompatibility(newDefinition);
-        newDefinition.checkCompatibility(thisDefinition);
-    }
+    protected abstract List<String> getKeyNames();
 
     protected static class ProcessPendingUpdater implements Updater {
         private BaseArrayBackedMutableTable baseArrayBackedMutableTable;
@@ -174,8 +166,8 @@ ArrayBackedMutableInputTable makeHandler() {
 
     protected class ArrayBackedMutableInputTable implements MutableInputTable {
         @Override
-        public InputTableDefinition getDefinition() {
-            return inputTableDefinition;
+        public List<String> getKeyNames() {
+            return BaseArrayBackedMutableTable.this.getKeyNames();
         }
 
         @Override
@@ -184,9 +176,13 @@ public TableDefinition getTableDefinition() {
         }
 
         @Override
-        public void add(Table newData) {
-            final long sequence = enqueueAddition(newData, true).sequence;
+        public void add(Table newData) throws IOException {
+            PendingChange pendingChange = enqueueAddition(newData, true);
+            final long sequence = pendingChange.sequence;
             waitForSequence(sequence);
+            if (pendingChange.error != null) {
+                throw new IOException(pendingChange.error);
+            }
         }
 
         private void add(Table newData, boolean allowEdits, InputTableStatusListener listener) {
@@ -204,7 +200,7 @@ private void add(Table newData, boolean allowEdits, InputTableStatusListener lis
         }
 
         private PendingChange enqueueAddition(Table newData, boolean allowEdits) {
-            validateDefinition(newData.getDefinition());
+            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 LTM refresh
             final PendingChange pendingChange = new PendingChange(doSnap(newData), false, allowEdits);
@@ -228,12 +224,16 @@ private Table doSnap(Table newData) {
         }
 
         @Override
-        public void delete(Table table, Index index) {
-            validateDelete(table.getDefinition());
+        public void delete(Table table, Index index) throws IOException {
+            validateDelete(table);
             final PendingChange pendingChange = new PendingChange(doSnap(table, index), true, false);
             pendingChanges.add(pendingChange);
             onPendingChange.run();
             waitForSequence(pendingChange.sequence);
+
+            if (pendingChange.error != null) {
+                throw new IOException(pendingChange.error);
+            }
         }
 
         @Override
diff --git a/DB/src/main/java/io/deephaven/db/v2/utils/KeyedArrayBackedMutableTable.java b/DB/src/main/java/io/deephaven/db/v2/utils/KeyedArrayBackedMutableTable.java
index 120f1306ce0..01222fa02d7 100644
--- a/DB/src/main/java/io/deephaven/db/v2/utils/KeyedArrayBackedMutableTable.java
+++ b/DB/src/main/java/io/deephaven/db/v2/utils/KeyedArrayBackedMutableTable.java
@@ -4,7 +4,6 @@
 import io.deephaven.db.exceptions.ArgumentException;
 import io.deephaven.db.tables.Table;
 import io.deephaven.db.tables.TableDefinition;
-import io.deephaven.db.tables.utils.TableTools;
 import io.deephaven.db.v2.QueryTable;
 import io.deephaven.db.v2.sources.*;
 import io.deephaven.db.v2.sources.chunk.*;
@@ -16,8 +15,6 @@
 
 import java.util.*;
 import java.util.function.Consumer;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 /**
  * An in-memory table that has keys for each row, which can be updated on the LTM.
@@ -27,7 +24,7 @@
 public class KeyedArrayBackedMutableTable extends BaseArrayBackedMutableTable {
     static final String DEFAULT_DESCRIPTION = "In-Memory Input Table";
 
-    private final String[] keyColumnNames;
+    private final List<String> keyColumnNames;
     private final Set<String> keyColumnSet;
     protected final ObjectArraySource<?>[] arrayValueSources;
 
@@ -109,15 +106,16 @@ private KeyedArrayBackedMutableTable(@NotNull TableDefinition definition, final
                     + ", available columns: " + definition.getColumnNames());
         }
 
-        this.keyColumnNames = keyColumnNames;
+        this.keyColumnNames = Collections.unmodifiableList(new ArrayList<>(Arrays.asList(keyColumnNames)));
         this.keyColumnSet = new HashSet<>(Arrays.asList(keyColumnNames));
-        inputTableDefinition.setKeys(keyColumnNames);
-        inputTableDefinition.setValues(
-                definition.getColumnNames().stream().filter(n -> !keyColumnSet.contains(n)).toArray(String[]::new));
-        final Stream<ObjectArraySource<?>> objectArraySourceStream =
-                Arrays.stream(inputTableDefinition.getValues()).map(this::getColumnSource)
-                        .filter(cs -> cs instanceof ObjectArraySource).map(cs -> (ObjectArraySource<?>) cs);
-        arrayValueSources = objectArraySourceStream.toArray(ObjectArraySource[]::new);
+        this.arrayValueSources =
+                definition.getColumnStream()
+                        .map(ColumnDefinition::getName)
+                        .filter(n -> !keyColumnSet.contains(n))
+                        .map(this::getColumnSource)
+                        .filter(cs -> cs instanceof ObjectArraySource)
+                        .map(cs -> (ObjectArraySource<?>) cs)
+                        .toArray(ObjectArraySource[]::new);
     }
 
     private void startTrackingPrev() {
@@ -237,7 +235,7 @@ protected void processPendingDelete(Table table, IndexChangeRecorder indexChange
     private ChunkSource<Attributes.Values> makeKeySource(Table table) {
         // noinspection unchecked
         return TupleSourceFactory.makeTupleSource(
-                Arrays.stream(keyColumnNames).map(table::getColumnSource).toArray(ColumnSource[]::new));
+                keyColumnNames.stream().map(table::getColumnSource).toArray(ColumnSource[]::new));
     }
 
     @Override
@@ -246,26 +244,8 @@ protected String getDefaultDescription() {
     }
 
     @Override
-    void validateDelete(final TableDefinition keyDefinition) {
-        final TableDefinition thisDefinition = getDefinition();
-        final StringBuilder error = new StringBuilder();
-        for (String keyColumn : keyColumnNames) {
-            final ColumnDefinition<?> colDef = keyDefinition.getColumn(keyColumn);
-            final ColumnDefinition<?> thisColDef = thisDefinition.getColumn(keyColumn);
-            if (colDef == null) {
-                error.append("Key Column \"").append(keyColumn).append("\" does not exist.\n");
-            } else if (!colDef.isCompatible(thisColDef)) {
-                error.append("Key Column \"").append(keyColumn).append("\" is not compatible.\n");
-            }
-        }
-        final List<String> extraKeys = keyDefinition.getColumnNames().stream().filter(kd -> !keyColumnSet.contains(kd))
-                .collect(Collectors.toList());
-        if (!extraKeys.isEmpty()) {
-            error.append("Unknown key columns: ").append(extraKeys);
-        }
-        if (error.length() > 0) {
-            throw new ArgumentException("Invalid Key Table Definition: " + error.toString());
-        }
+    protected List<String> getKeyNames() {
+        return keyColumnNames;
     }
 
     /**
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 960aa696837..9407e12e76f 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
@@ -159,6 +159,7 @@ public static int makeSchemaPayload(final FlatBufferBuilder builder,
         final Map<String, String> schemaMetadata = new HashMap<>();
 
         // copy primitives as strings
+        Set<String> unsentAttributes = new HashSet<>();
         for (final Map.Entry<String, Object> entry : attributes.entrySet()) {
             final String key = entry.getKey();
             final Object val = entry.getValue();
@@ -167,11 +168,14 @@ public static int makeSchemaPayload(final FlatBufferBuilder builder,
                     val instanceof Character || val instanceof Boolean ||
                     (val instanceof String && ((String) val).length() < ATTR_STRING_LEN_CUTOFF)) {
                 putMetadata(schemaMetadata, "attribute." + key, val.toString());
+            } else {
+                unsentAttributes.add(key);
             }
         }
 
         // copy rollup details
         if (attributes.containsKey(Table.HIERARCHICAL_SOURCE_INFO_ATTRIBUTE)) {
+            unsentAttributes.remove(Table.HIERARCHICAL_SOURCE_INFO_ATTRIBUTE);
             final HierarchicalTableInfo hierarchicalTableInfo =
                     (HierarchicalTableInfo) attributes.remove(Table.HIERARCHICAL_SOURCE_INFO_ATTRIBUTE);
             final String hierarchicalSourceKeyPrefix = "attribute." + Table.HIERARCHICAL_SOURCE_INFO_ATTRIBUTE + ".";
@@ -190,6 +194,11 @@ public static int makeSchemaPayload(final FlatBufferBuilder builder,
             }
         }
 
+        // note which attributes have a value we couldn't send
+        for (String unsentAttribute : unsentAttributes) {
+            putMetadata(schemaMetadata, "unsent.attribute." + unsentAttribute, "");
+        }
+
         final Map<String, Field> fields = new LinkedHashMap<>();
         for (final ColumnDefinition<?> column : table.getColumns()) {
             final String colName = column.getName();
@@ -437,7 +446,7 @@ private static Field arrowFieldFor(final String name, final ColumnDefinition<?>
             putMetadata(metadata, "description", description);
         }
         if (inputTable != null) {
-            putMetadata(metadata, "inputtable.isKey", Arrays.asList(inputTable.getKeyNames()).contains(name) + "");
+            putMetadata(metadata, "inputtable.isKey", inputTable.getKeyNames().contains(name) + "");
         }
 
         return arrowFieldFor(name, type, componentType, metadata);
diff --git a/grpc-api/src/flightTest/java/io/deephaven/grpc_api/flight/FlightMessageRoundTripTest.java b/grpc-api/src/flightTest/java/io/deephaven/grpc_api/flight/FlightMessageRoundTripTest.java
index c409fd7751f..f7be299207a 100644
--- a/grpc-api/src/flightTest/java/io/deephaven/grpc_api/flight/FlightMessageRoundTripTest.java
+++ b/grpc-api/src/flightTest/java/io/deephaven/grpc_api/flight/FlightMessageRoundTripTest.java
@@ -416,7 +416,7 @@ private void assertSchemaMatchesTable(Schema schema, Table table) {
         Assert.eq(schema.getFields().size(), "schema.getFields().size()", table.getColumns().length,
                 "table.getColumns().length");
         Assert.equals(BarrageUtil.convertArrowSchema(schema).tableDef,
-                "BarrageSchemaUtil.schemaToTableDefinition(schema)",
+                "BarrageUtil.convertArrowSchema(schema)",
                 table.getDefinition(), "table.getDefinition()");
     }
 
diff --git a/grpc-api/src/main/java/io/deephaven/grpc_api/runner/DeephavenApiServerModule.java b/grpc-api/src/main/java/io/deephaven/grpc_api/runner/DeephavenApiServerModule.java
index d6cfd836446..c90d5569933 100644
--- a/grpc-api/src/main/java/io/deephaven/grpc_api/runner/DeephavenApiServerModule.java
+++ b/grpc-api/src/main/java/io/deephaven/grpc_api/runner/DeephavenApiServerModule.java
@@ -14,6 +14,7 @@
 import io.deephaven.grpc_api.log.LogModule;
 import io.deephaven.grpc_api.session.SessionModule;
 import io.deephaven.grpc_api.table.TableModule;
+import io.deephaven.grpc_api.table.inputtables.InputTableModule;
 import io.deephaven.grpc_api.uri.UriModule;
 import io.deephaven.grpc_api.util.Scheduler;
 import io.deephaven.util.process.ProcessEnvironment;
@@ -46,6 +47,7 @@
         UriModule.class,
         SessionModule.class,
         TableModule.class,
+        InputTableModule.class,
         ConsoleModule.class,
         GroovyConsoleSessionModule.class,
         PythonConsoleSessionModule.class
diff --git a/grpc-api/src/main/java/io/deephaven/grpc_api/table/TableModule.java b/grpc-api/src/main/java/io/deephaven/grpc_api/table/TableModule.java
index 7d3db71e6ec..db0bb12a6df 100644
--- a/grpc-api/src/main/java/io/deephaven/grpc_api/table/TableModule.java
+++ b/grpc-api/src/main/java/io/deephaven/grpc_api/table/TableModule.java
@@ -7,6 +7,7 @@
 import dagger.multibindings.IntoSet;
 import io.deephaven.grpc_api.table.ops.ApplyPreviewColumnsGrpcImpl;
 import io.deephaven.grpc_api.table.ops.ComboAggregateGrpcImpl;
+import io.deephaven.grpc_api.table.ops.CreateInputTableGrpcImpl;
 import io.deephaven.grpc_api.table.ops.DropColumnsGrpcImpl;
 import io.deephaven.grpc_api.table.ops.EmptyTableGrpcImpl;
 import io.deephaven.grpc_api.table.ops.FetchTableGrpcImpl;
@@ -184,4 +185,9 @@ public interface TableModule {
     @IntoMap
     @BatchOpCode(BatchTableRequest.Operation.OpCase.APPLY_PREVIEW_COLUMNS)
     GrpcTableOperation<?> bindApplyPreviewColumns(ApplyPreviewColumnsGrpcImpl op);
+
+    @Binds
+    @IntoMap
+    @BatchOpCode(BatchTableRequest.Operation.OpCase.CREATE_INPUT_TABLE)
+    GrpcTableOperation<?> bindCreateInputTable(CreateInputTableGrpcImpl op);
 }
diff --git a/grpc-api/src/main/java/io/deephaven/grpc_api/table/TableServiceGrpcImpl.java b/grpc-api/src/main/java/io/deephaven/grpc_api/table/TableServiceGrpcImpl.java
index 24ffe90b085..fd0dbc14ae5 100644
--- a/grpc-api/src/main/java/io/deephaven/grpc_api/table/TableServiceGrpcImpl.java
+++ b/grpc-api/src/main/java/io/deephaven/grpc_api/table/TableServiceGrpcImpl.java
@@ -19,6 +19,7 @@
 import io.deephaven.proto.backplane.grpc.AsOfJoinTablesRequest;
 import io.deephaven.proto.backplane.grpc.BatchTableRequest;
 import io.deephaven.proto.backplane.grpc.ComboAggregateRequest;
+import io.deephaven.proto.backplane.grpc.CreateInputTableRequest;
 import io.deephaven.proto.backplane.grpc.CrossJoinTablesRequest;
 import io.deephaven.proto.backplane.grpc.DropColumnsRequest;
 import io.deephaven.proto.backplane.grpc.EmptyTableRequest;
@@ -259,6 +260,12 @@ public void applyPreviewColumns(ApplyPreviewColumnsRequest request,
         oneShotOperationWrapper(BatchTableRequest.Operation.OpCase.APPLY_PREVIEW_COLUMNS, request, responseObserver);
     }
 
+    @Override
+    public void createInputTable(CreateInputTableRequest request,
+            StreamObserver<ExportedTableCreationResponse> responseObserver) {
+        oneShotOperationWrapper(BatchTableRequest.Operation.OpCase.CREATE_INPUT_TABLE, request, responseObserver);
+    }
+
     @Override
     public void batch(final BatchTableRequest request,
             final StreamObserver<ExportedTableCreationResponse> responseObserver) {
diff --git a/grpc-api/src/main/java/io/deephaven/grpc_api/table/inputtables/InputTableModule.java b/grpc-api/src/main/java/io/deephaven/grpc_api/table/inputtables/InputTableModule.java
new file mode 100644
index 00000000000..d0bf941e8c0
--- /dev/null
+++ b/grpc-api/src/main/java/io/deephaven/grpc_api/table/inputtables/InputTableModule.java
@@ -0,0 +1,13 @@
+package io.deephaven.grpc_api.table.inputtables;
+
+import dagger.Binds;
+import dagger.Module;
+import dagger.multibindings.IntoSet;
+import io.grpc.BindableService;
+
+@Module
+public interface InputTableModule {
+    @Binds
+    @IntoSet
+    BindableService bindInputTableServiceGrpcImpl(InputTableServiceGrpcImpl inputTableService);
+}
diff --git a/grpc-api/src/main/java/io/deephaven/grpc_api/table/inputtables/InputTableServiceGrpcImpl.java b/grpc-api/src/main/java/io/deephaven/grpc_api/table/inputtables/InputTableServiceGrpcImpl.java
new file mode 100644
index 00000000000..03e5f80b7f3
--- /dev/null
+++ b/grpc-api/src/main/java/io/deephaven/grpc_api/table/inputtables/InputTableServiceGrpcImpl.java
@@ -0,0 +1,121 @@
+package io.deephaven.grpc_api.table.inputtables;
+
+import com.google.rpc.Code;
+import io.deephaven.db.tables.Table;
+import io.deephaven.db.tables.TableDefinition;
+import io.deephaven.db.util.config.MutableInputTable;
+import io.deephaven.extensions.barrage.util.GrpcUtil;
+import io.deephaven.grpc_api.session.SessionService;
+import io.deephaven.grpc_api.session.SessionState;
+import io.deephaven.grpc_api.session.TicketRouter;
+import io.deephaven.internal.log.LoggerFactory;
+import io.deephaven.io.logger.Logger;
+import io.deephaven.proto.backplane.grpc.*;
+import io.grpc.stub.StreamObserver;
+
+import javax.inject.Inject;
+import java.io.IOException;
+
+public class InputTableServiceGrpcImpl extends InputTableServiceGrpc.InputTableServiceImplBase {
+
+    private static final Logger log = LoggerFactory.getLogger(InputTableServiceGrpcImpl.class);
+
+    private final TicketRouter ticketRouter;
+    private final SessionService sessionService;
+
+    @Inject
+    public InputTableServiceGrpcImpl(TicketRouter ticketRouter, SessionService sessionService) {
+        this.ticketRouter = ticketRouter;
+        this.sessionService = sessionService;
+    }
+
+    @Override
+    public void addTableToInputTable(AddTableRequest request, StreamObserver<AddTableResponse> responseObserver) {
+        GrpcUtil.rpcWrapper(log, responseObserver, () -> {
+            final SessionState session = sessionService.getCurrentSession();
+
+            SessionState.ExportObject<Table> targetTable =
+                    ticketRouter.resolve(session, request.getInputTable(), "inputTable");
+            SessionState.ExportObject<Table> tableToAdd =
+                    ticketRouter.resolve(session, request.getTableToAdd(), "tableToAdd");
+
+            session.nonExport()
+                    .requiresSerialQueue()
+                    .onError(responseObserver)
+                    .require(targetTable, tableToAdd)
+                    .submit(() -> {
+                        Object inputTable = targetTable.get().getAttribute(Table.INPUT_TABLE_ATTRIBUTE);
+                        if (!(inputTable instanceof MutableInputTable)) {
+                            throw GrpcUtil.statusRuntimeException(Code.INVALID_ARGUMENT,
+                                    "Table can't be used as an input table");
+                        }
+
+                        MutableInputTable mutableInputTable = (MutableInputTable) inputTable;
+                        Table table = tableToAdd.get();
+
+                        // validate that the columns are compatible
+                        try {
+                            mutableInputTable.validateAddOrModify(table);
+                        } catch (TableDefinition.IncompatibleTableDefinitionException exception) {
+                            throw GrpcUtil.statusRuntimeException(Code.INVALID_ARGUMENT,
+                                    "Provided tables's columns are not compatible: " + exception.getMessage());
+                        }
+
+                        // actually add the tables contents
+                        try {
+                            mutableInputTable.add(table);
+                        } catch (IOException ioException) {
+                            throw GrpcUtil.statusRuntimeException(Code.DATA_LOSS, "Error adding table to input table");
+                        }
+                    });
+        });
+    }
+
+    @Override
+    public void deleteTableFromInputTable(DeleteTableRequest request,
+            StreamObserver<DeleteTableResponse> responseObserver) {
+        GrpcUtil.rpcWrapper(log, responseObserver, () -> {
+            final SessionState session = sessionService.getCurrentSession();
+
+            SessionState.ExportObject<Table> targetTable =
+                    ticketRouter.resolve(session, request.getInputTable(), "inputTable");
+            SessionState.ExportObject<Table> tableToDeleteExport =
+                    ticketRouter.resolve(session, request.getTableToRemove(), "tableToDelete");
+
+            session.nonExport()
+                    .requiresSerialQueue()
+                    .onError(responseObserver)
+                    .require(targetTable, tableToDeleteExport)
+                    .submit(() -> {
+                        Object inputTable = targetTable.get().getAttribute(Table.INPUT_TABLE_ATTRIBUTE);
+                        if (!(inputTable instanceof MutableInputTable)) {
+                            throw GrpcUtil.statusRuntimeException(Code.INVALID_ARGUMENT,
+                                    "Table can't be used as an input table");
+                        }
+
+                        MutableInputTable mutableInputTable = (MutableInputTable) inputTable;
+
+                        Table tableToDelete = tableToDeleteExport.get();
+
+                        // validate that the columns are compatible
+                        try {
+                            mutableInputTable.validateDelete(tableToDelete);
+                        } catch (TableDefinition.IncompatibleTableDefinitionException exception) {
+                            throw GrpcUtil.statusRuntimeException(Code.INVALID_ARGUMENT,
+                                    "Provided tables's columns are not compatible: " + exception.getMessage());
+                        } catch (UnsupportedOperationException exception) {
+                            throw GrpcUtil.statusRuntimeException(Code.INVALID_ARGUMENT,
+                                    "Provided input table does not support delete.");
+                        }
+
+                        // actually delete the table's contents
+                        try {
+                            mutableInputTable.delete(tableToDelete);
+                        } catch (IOException ioException) {
+                            throw GrpcUtil.statusRuntimeException(Code.DATA_LOSS,
+                                    "Error deleting table from inputtable");
+                        }
+                    });
+        });
+    }
+}
diff --git a/grpc-api/src/main/java/io/deephaven/grpc_api/table/ops/CreateInputTableGrpcImpl.java b/grpc-api/src/main/java/io/deephaven/grpc_api/table/ops/CreateInputTableGrpcImpl.java
new file mode 100644
index 00000000000..f35b7a96a34
--- /dev/null
+++ b/grpc-api/src/main/java/io/deephaven/grpc_api/table/ops/CreateInputTableGrpcImpl.java
@@ -0,0 +1,79 @@
+package io.deephaven.grpc_api.table.ops;
+
+import com.google.rpc.Code;
+import io.deephaven.datastructures.util.CollectionUtil;
+import io.deephaven.db.tables.Table;
+import io.deephaven.db.tables.TableDefinition;
+import io.deephaven.db.v2.utils.AppendOnlyArrayBackedMutableTable;
+import io.deephaven.db.v2.utils.KeyedArrayBackedMutableTable;
+import io.deephaven.extensions.barrage.util.BarrageUtil;
+import io.deephaven.extensions.barrage.util.GrpcUtil;
+import io.deephaven.grpc_api.session.SessionState;
+import io.deephaven.grpc_api.util.SchemaHelper;
+import io.deephaven.proto.backplane.grpc.BatchTableRequest;
+import io.deephaven.proto.backplane.grpc.CreateInputTableRequest;
+import io.grpc.StatusRuntimeException;
+import org.apache.arrow.flatbuf.Schema;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+import java.util.Collections;
+import java.util.List;
+
+import static io.deephaven.proto.backplane.grpc.CreateInputTableRequest.InputTableKind.KindCase.KIND_NOT_SET;
+
+@Singleton
+public class CreateInputTableGrpcImpl extends GrpcTableOperation<CreateInputTableRequest> {
+
+    private static final MultiDependencyFunction<CreateInputTableRequest> optionalSourceTable =
+            (CreateInputTableRequest req) -> req.hasSourceTableId()
+                    ? Collections.singletonList(req.getSourceTableId())
+                    : Collections.emptyList();
+
+    @Inject
+    public CreateInputTableGrpcImpl() {
+        super(BatchTableRequest.Operation::getCreateInputTable, CreateInputTableRequest::getResultId,
+                optionalSourceTable);
+    }
+
+    @Override
+    public void validateRequest(CreateInputTableRequest request) throws StatusRuntimeException {
+        // ensure we have one of either schema or source table (protobuf will ensure we don't have both)
+        if (!request.hasSchema() && !request.hasSourceTableId()) {
+            throw GrpcUtil.statusRuntimeException(Code.INVALID_ARGUMENT,
+                    "Must specify one of schema and source_table_id");
+        }
+
+        if (request.getKind().getKindCase() == null ||
+                request.getKind().getKindCase() == KIND_NOT_SET) {
+            throw GrpcUtil.statusRuntimeException(Code.INVALID_ARGUMENT,
+                    "Unrecognized InputTableKind");
+        }
+    }
+
+    @Override
+    public Table create(CreateInputTableRequest request, List<SessionState.ExportObject<Table>> sourceTables) {
+        TableDefinition tableDefinitionFromSchema;
+
+        if (request.hasSchema()) {
+            Schema schema = SchemaHelper.flatbufSchema(request.getSchema().asReadOnlyByteBuffer());
+            tableDefinitionFromSchema = BarrageUtil.convertArrowSchema(schema).tableDef;
+        } else if (request.hasSourceTableId()) {
+            Table sourceTable = sourceTables.get(0).get();
+            tableDefinitionFromSchema = sourceTable.getDefinition();
+        } else {
+            throw new IllegalStateException("missing schema and source_table_id");
+        }
+        switch (request.getKind().getKindCase()) {
+            case IN_MEMORY_APPEND_ONLY:
+                return AppendOnlyArrayBackedMutableTable.make(tableDefinitionFromSchema);
+            case IN_MEMORY_KEY_BACKED:
+                return KeyedArrayBackedMutableTable.make(tableDefinitionFromSchema,
+                        request.getKind().getInMemoryKeyBacked().getKeyColumnsList()
+                                .toArray(CollectionUtil.ZERO_LENGTH_STRING_ARRAY));
+            case KIND_NOT_SET:
+            default:
+                throw new IllegalStateException("Unsupported input table kind");
+        }
+    }
+}
diff --git a/proto/proto-backplane-grpc-flight/src/main/java/io/deephaven/grpc_api/util/SchemaHelper.java b/proto/proto-backplane-grpc-flight/src/main/java/io/deephaven/grpc_api/util/SchemaHelper.java
index 30820acd4c0..befdb742de1 100644
--- a/proto/proto-backplane-grpc-flight/src/main/java/io/deephaven/grpc_api/util/SchemaHelper.java
+++ b/proto/proto-backplane-grpc-flight/src/main/java/io/deephaven/grpc_api/util/SchemaHelper.java
@@ -28,7 +28,16 @@ public static Schema schema(ExportedTableCreationResponse response) {
      * @return the flatbuf schema
      */
     public static org.apache.arrow.flatbuf.Schema flatbufSchema(ExportedTableCreationResponse response) {
-        final ByteBuffer bb = response.getSchemaHeader().asReadOnlyByteBuffer();
+        return flatbufSchema(response.getSchemaHeader().asReadOnlyByteBuffer());
+    }
+
+    /**
+     * Creates a flatbuf Schema from raw bytes of a Message.
+     * 
+     * @param bb a bytebuffer that contains a schema in a message
+     * @return a flatbuf schema
+     */
+    public static org.apache.arrow.flatbuf.Schema flatbufSchema(ByteBuffer bb) {
         if (bb.remaining() < MESSAGE_OFFSET) {
             throw new IllegalArgumentException("Not enough bytes for Message/Schema");
         }
diff --git a/proto/proto-backplane-grpc/Dockerfile b/proto/proto-backplane-grpc/Dockerfile
index b6e85a217cc..17d72696b41 100644
--- a/proto/proto-backplane-grpc/Dockerfile
+++ b/proto/proto-backplane-grpc/Dockerfile
@@ -20,7 +20,8 @@ RUN set -eux; \
     /includes/deephaven/proto/console.proto \
     /includes/deephaven/proto/session.proto \
     /includes/deephaven/proto/table.proto \
-    /includes/deephaven/proto/application.proto; \
+    /includes/deephaven/proto/application.proto \
+    /includes/deephaven/proto/inputtable.proto; \
   /opt/protoc/bin/protoc \
     --plugin=protoc-gen-ts=/usr/src/app/node_modules/.bin/protoc-gen-ts \
     --js_out=import_style=commonjs:/generated/js \
@@ -33,7 +34,8 @@ RUN set -eux; \
     /includes/deephaven/proto/console.proto \
     /includes/deephaven/proto/session.proto \
     /includes/deephaven/proto/table.proto \
-    /includes/deephaven/proto/application.proto; \
+    /includes/deephaven/proto/application.proto \
+    /includes/deephaven/proto/inputtable.proto; \
   python3 -m grpc_tools.protoc \
     --grpc_python_out=/generated/python \
     --python_out=/generated/python \
@@ -42,4 +44,5 @@ RUN set -eux; \
     /includes/deephaven/proto/console.proto \
     /includes/deephaven/proto/session.proto \
     /includes/deephaven/proto/table.proto \
-    /includes/deephaven/proto/application.proto;
+    /includes/deephaven/proto/application.proto \
+    /includes/deephaven/proto/inputtable.proto;
diff --git a/proto/proto-backplane-grpc/src/main/java/io/deephaven/grpc_api/util/OperationHelper.java b/proto/proto-backplane-grpc/src/main/java/io/deephaven/grpc_api/util/OperationHelper.java
index 0a459f4d2e2..a411f5b92af 100644
--- a/proto/proto-backplane-grpc/src/main/java/io/deephaven/grpc_api/util/OperationHelper.java
+++ b/proto/proto-backplane-grpc/src/main/java/io/deephaven/grpc_api/util/OperationHelper.java
@@ -73,6 +73,10 @@ public static Stream<TableReference> getSourceIds(Operation op) {
                 return Stream.of(op.getFetchPandasTable().getSourceId());
             case APPLY_PREVIEW_COLUMNS:
                 return Stream.of(op.getApplyPreviewColumns().getSourceId());
+            case CREATE_INPUT_TABLE:
+                return op.getCreateInputTable().hasSourceTableId()
+                        ? Stream.of(op.getCreateInputTable().getSourceTableId())
+                        : Stream.empty();
             case OP_NOT_SET:
                 throw new IllegalStateException("Operation id not set");
             default:
diff --git a/proto/proto-backplane-grpc/src/main/proto/deephaven/proto/inputtable.proto b/proto/proto-backplane-grpc/src/main/proto/deephaven/proto/inputtable.proto
new file mode 100644
index 00000000000..2e97fd1f44b
--- /dev/null
+++ b/proto/proto-backplane-grpc/src/main/proto/deephaven/proto/inputtable.proto
@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) 2016-2021 Deephaven Data Labs and Patent Pending
+ */
+
+syntax = "proto3";
+
+package io.deephaven.proto.backplane.grpc;
+
+option java_multiple_files = true;
+option optimize_for = SPEED;
+
+import "deephaven/proto/ticket.proto";
+
+/*
+ * This service offers methods to manipulate the contents of input tables.
+ */
+service InputTableService {
+    /*
+     * Adds the provided table to the specified input table. The new data to add must only have
+     * columns (name, types, and order) which match the given input table's columns.
+     */
+    rpc AddTableToInputTable(AddTableRequest) returns (AddTableResponse) {}
+
+    /*
+    * Removes the provided table from the specified input tables. The tables indicating which rows
+    * to remove are expected to only have columns that match the key columns of the input table.
+    */
+    rpc DeleteTableFromInputTable(DeleteTableRequest) returns (DeleteTableResponse) {}
+
+}
+
+message AddTableRequest {
+    Ticket input_table = 1;
+    Ticket table_to_add = 2;
+}
+
+message AddTableResponse {
+
+}
+
+message DeleteTableRequest {
+    Ticket input_table = 1;
+    Ticket table_to_remove = 2;
+}
+
+message DeleteTableResponse {
+
+}
\ No newline at end of file
diff --git a/proto/proto-backplane-grpc/src/main/proto/deephaven/proto/table.proto b/proto/proto-backplane-grpc/src/main/proto/deephaven/proto/table.proto
index 98ada4d4467..5dcdb395b94 100644
--- a/proto/proto-backplane-grpc/src/main/proto/deephaven/proto/table.proto
+++ b/proto/proto-backplane-grpc/src/main/proto/deephaven/proto/table.proto
@@ -178,6 +178,12 @@ service TableService {
    */
   rpc RunChartDownsample(RunChartDownsampleRequest) returns (ExportedTableCreationResponse) {}
 
+  /**
+   * Creates a new Table based on the provided configuration. This can be used as a regular table from the other methods
+   * in this interface, or can be interacted with via the InputTableService to modify its contents.
+   */
+  rpc CreateInputTable(CreateInputTableRequest) returns (ExportedTableCreationResponse) {}
+
   /*
    * Batch a series of requests and send them all at once. This enables the user to create intermediate tables without
    * requiring them to be exported and managed by the client. The server will automatically release any tables when they
@@ -579,6 +585,34 @@ message RunChartDownsampleRequest {
   repeated string y_column_names = 6;
 }
 
+message CreateInputTableRequest {
+  message InputTableKind {
+    // Creates an in-memory append-only table - rows cannot be modified or deleted.
+    message InMemoryAppendOnly {
+
+    }
+    // Creates an in-memory table that supports updates and deletes by keys.
+    message InMemoryKeyBacked {
+      repeated string key_columns = 1;
+    }
+    oneof kind {
+      InMemoryAppendOnly in_memory_append_only = 1;
+      InMemoryKeyBacked in_memory_key_backed = 2;
+    }
+  }
+
+  Ticket result_id = 1;
+  oneof definition {
+    // Optional, either this or schema must be specified, not both.
+    TableReference source_table_id = 2;
+    // Schema as described in Arrow Message.fbs::Message. Optional, either this or source_table_id must be specified.
+    bytes schema = 3;
+  }
+
+  // Specifies what type of input table to create.
+  InputTableKind kind = 4;
+}
+
 message BatchTableRequest {
   repeated Operation ops = 1;
 
@@ -614,6 +648,7 @@ message BatchTableRequest {
       FetchTableRequest fetch_table = 28;
       FetchPandasTableRequest fetch_pandas_table = 29;
       ApplyPreviewColumnsRequest apply_preview_columns = 30;
+      CreateInputTableRequest create_input_table = 31;
     }
   }
 }
diff --git a/web/shared-beans/src/main/java/io/deephaven/web/shared/data/InputTableDefinition.java b/web/shared-beans/src/main/java/io/deephaven/web/shared/data/InputTableDefinition.java
deleted file mode 100644
index 7d9398757c4..00000000000
--- a/web/shared-beans/src/main/java/io/deephaven/web/shared/data/InputTableDefinition.java
+++ /dev/null
@@ -1,28 +0,0 @@
-package io.deephaven.web.shared.data;
-
-import java.io.Serializable;
-
-/**
- * Adds extra information necessary to treat a table as an InputTable instance.
- */
-public class InputTableDefinition implements Serializable {
-
-    private String[] keys;
-    private String[] values;
-
-    public String[] getKeys() {
-        return keys;
-    }
-
-    public void setKeys(String[] keys) {
-        this.keys = keys;
-    }
-
-    public String[] getValues() {
-        return values;
-    }
-
-    public void setValues(String[] values) {
-        this.values = values;
-    }
-}