[ZEPPELIN-753] TableData abstraction

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterResultMessage.java

@@ -16,10 +16,12 @@
  */
 package org.apache.zeppelin.interpreter;
 
+import java.io.Serializable;
+
 /**
  * Interpreter result message
  */
-public class InterpreterResultMessage {
+public class InterpreterResultMessage implements Serializable {
   InterpreterResult.Type type;
   String data;
 

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java

@@ -351,7 +351,6 @@ public void putResponseInvokeMethod(
     }
   }
 
-
   /**
    * Supposed to call from RemoteInterpreterEventPoller
    * @return next available event

zeppelin-interpreter/src/main/java/org/apache/zeppelin/resource/Resource.java

@@ -117,6 +117,7 @@ public Object invokeMethod(
         Method method = r.getClass().getMethod(
             methodName,
             paramTypes);
+        method.setAccessible(true);
         Object ret = method.invoke(r, params);
         return ret;
       }  catch (Exception e) {

zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/ColumnDef.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zeppelin.tabledata;
+
+import java.io.Serializable;
+
+/**
+ * Column definition
+ */
+public class ColumnDef implements Serializable {
+  /**
+   * Type
+   */
+  public static enum TYPE {
+    STRING,
+    LONG,
+    INT
+  }
+
+  private String name;
+  private TYPE type;
+
+  public ColumnDef(String name, TYPE type) {
+    this.name = name;
+    this.type = type;
+  }
+
+  public String name() {
+    return name;
+  }
+
+  public TYPE type() {
+    return type;
+  }
+}

zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/InterpreterResultTableData.java

@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zeppelin.tabledata;
+
+import org.apache.zeppelin.interpreter.InterpreterResultMessage;
+
+import java.io.Serializable;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Table data with interpreter result type 'TABLE'
+ */
+public class InterpreterResultTableData implements TableData, Serializable {
+  private final InterpreterResultMessage msg;
+  ColumnDef [] columnDef;
+  List<Row> rows = new LinkedList<>();
+
+  public InterpreterResultTableData(InterpreterResultMessage msg) {
+    this.msg = msg;
+
+    String[] lines = msg.getData().split("\n");
+    if (lines == null || lines.length == 0) {
+      columnDef = null;
+    } else {
+      String[] headerRow = lines[0].split("\t");
+      columnDef = new ColumnDef[headerRow.length];
+      for (int i = 0; i < headerRow.length; i++) {
+        columnDef[i] = new ColumnDef(headerRow[i], ColumnDef.TYPE.STRING);
+      }
+
+      for (int r = 1; r < lines.length; r++) {
+        Object [] row = lines[r].split("\t");
+        rows.add(new Row(row));
+      }
+    }
+  }
+
+
+  @Override
+  public ColumnDef[] columns() {
+    return columnDef;
+  }
+
+  @Override
+  public Iterator<Row> rows() {
+    return rows.iterator();
+  }
+}

zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/ProxyRowIterator.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zeppelin.tabledata;
+
+import org.apache.zeppelin.resource.Resource;
+
+import java.util.Iterator;
+
+/**
+ * Proxy row iterator
+ */
+public class ProxyRowIterator implements Iterator<Row> {
+
+  private final Resource rows;
+
+  public ProxyRowIterator(Resource rows) {
+    this.rows = rows;
+  }
+
+  @Override
+  public boolean hasNext() {
+    rows.invokeMethod("hasNext", null, null);
+    return false;
+  }
+
+  @Override
+  public Row next() {
+    return (Row) rows.invokeMethod("next", null, null);
+  }
+
+  @Override
+  public void remove() {
+    // operation not supported
+  }
+}

zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/Row.java

@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zeppelin.tabledata;
+
+import java.io.Serializable;
+
+/**
+ * Row representation of table data
+ */
+public class Row implements Serializable {
+  private final Object[] data;
+
+  public Row(Object [] data) {
+    this.data = data;
+  }
+
+  public Object [] get() {
+    return data;
+  }
+}

zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/TableData.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zeppelin.tabledata;
+
+import java.util.Iterator;
+
+/**
+ * Abstract representation of table data
+ */
+public interface TableData {
+  /**
+   * Get column definitions
+   * @return
+   */
+  public ColumnDef [] columns();
+
+  /**
+   * Get row iterator
+   * @param
+   * @return
+   */
+  public Iterator<Row> rows();
+}

zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/TableDataException.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zeppelin.tabledata;
+
+import java.io.IOException;
+
+/**
+ * TableDataException
+ */
+public class TableDataException extends IOException {
+  public TableDataException(String s) {
+    super(s);
+  }
+}

zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/TableDataProxy.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zeppelin.tabledata;
+
+import org.apache.zeppelin.resource.Resource;
+import org.apache.zeppelin.resource.ResourcePoolUtils;
+
+import java.util.Iterator;
+
+/**
+ * Proxy TableData for ResourcePool
+ */
+public class TableDataProxy implements TableData {
+  private final Resource resource;
+
+  public TableDataProxy(Resource tableDataRemoteResource) {
+    this.resource = tableDataRemoteResource;
+  }
+
+  @Override
+  public ColumnDef[] columns() {
+    return (ColumnDef[]) resource.invokeMethod(
+        "columns", null, null);
+  }
+
+  @Override
+  public Iterator<Row> rows() {
+    String resourceName = resource.getResourceId().getName() + ".rows";
+    Resource rows = resource.invokeMethod("rows", null, null, resourceName);
+
+    ProxyRowIterator it = new ProxyRowIterator(rows);
+    return it;
+  }
+}

zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterResourcePool.java

@@ -133,4 +133,4 @@ public int getProgress(InterpreterContext context) {
   public List<InterpreterCompletion> completion(String buf, int cursor) {
     return null;
   }
-}
+}