[2/4] beam git commit: [BEAM-2528] create table

[mi...@apache.org]

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/Column.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/Column.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/Column.java
new file mode 100644
index 0000000..9bcc16a
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/Column.java
@@ -0,0 +1,51 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta;
+
+import com.google.auto.value.AutoValue;
+import java.io.Serializable;
+import javax.annotation.Nullable;
+
+/**
+ * Metadata class for a {@code BeamSqlTable} column.
+ */
+@AutoValue
+public abstract class Column implements Serializable {
+  public abstract String getName();
+  public abstract Integer getType();
+  @Nullable
+  public abstract String getComment();
+  public abstract boolean isPrimaryKey();
+
+  public static Builder builder() {
+    return new org.apache.beam.sdk.extensions.sql.meta.AutoValue_Column.Builder();
+  }
+
+  /**
+   * Builder class for {@link Column}.
+   */
+  @AutoValue.Builder
+  public abstract static class Builder {
+    public abstract Builder name(String name);
+    public abstract Builder type(Integer type);
+    public abstract Builder comment(String comment);
+    public abstract Builder primaryKey(boolean isPrimaryKey);
+    public abstract Column build();
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/Table.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/Table.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/Table.java
new file mode 100644
index 0000000..4af82a0
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/Table.java
@@ -0,0 +1,69 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta;
+
+import com.alibaba.fastjson.JSONObject;
+import com.google.auto.value.AutoValue;
+import java.io.Serializable;
+import java.net.URI;
+import java.util.List;
+import javax.annotation.Nullable;
+
+/**
+ * Represents the metadata of a {@code BeamSqlTable}.
+ */
+@AutoValue
+public abstract class Table implements Serializable {
+  /** type of the table. */
+  public abstract String getType();
+  public abstract String getName();
+  public abstract List<Column> getColumns();
+  @Nullable
+  public abstract String getComment();
+  @Nullable
+  public abstract URI getLocation();
+  @Nullable
+  public abstract JSONObject getProperties();
+
+  public static Builder builder() {
+    return new org.apache.beam.sdk.extensions.sql.meta.AutoValue_Table.Builder();
+  }
+
+  public String getLocationAsString() {
+    if (getLocation() == null) {
+      return null;
+    }
+
+    return "/" + getLocation().getHost() + getLocation().getPath();
+  }
+
+  /**
+   * Builder class for {@link Table}.
+   */
+  @AutoValue.Builder
+  public abstract static class Builder {
+    public abstract Builder type(String type);
+    public abstract Builder name(String name);
+    public abstract Builder columns(List<Column> columns);
+    public abstract Builder comment(String name);
+    public abstract Builder location(URI location);
+    public abstract Builder properties(JSONObject properties);
+    public abstract Table build();
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/package-info.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/package-info.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/package-info.java
new file mode 100644
index 0000000..a50ce5f
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * Metadata related classes.
+ */
+package org.apache.beam.sdk.extensions.sql.meta;

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/MetaUtils.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/MetaUtils.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/MetaUtils.java
new file mode 100644
index 0000000..35ecdce
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/MetaUtils.java
@@ -0,0 +1,40 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.provider;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.extensions.sql.meta.Column;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+
+/**
+ * Utility methods for metadata.
+ */
+public class MetaUtils {
+  public static BeamRecordSqlType getBeamSqlRecordTypeFromTable(Table table) {
+    List<String> columnNames = new ArrayList<>(table.getColumns().size());
+    List<Integer> columnTypes = new ArrayList<>(table.getColumns().size());
+    for (Column column : table.getColumns()) {
+      columnNames.add(column.getName());
+      columnTypes.add(column.getType());
+    }
+    return BeamRecordSqlType.create(columnNames, columnTypes);
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/TableProvider.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/TableProvider.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/TableProvider.java
new file mode 100644
index 0000000..d57f703
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/TableProvider.java
@@ -0,0 +1,61 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.provider;
+
+import java.util.List;
+import org.apache.beam.sdk.extensions.sql.BeamSqlTable;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+
+/**
+ * A {@code TableProvider} handles the metadata CRUD of a specified kind of tables.
+ *
+ * <p>So there will be a provider to handle textfile(CSV) based tables, there is a provider to
+ * handle MySQL based tables, a provider to handle Casandra based tables etc.
+ */
+public interface TableProvider {
+  /**
+   * Init the provider.
+   */
+  void init();
+
+  /**
+   * Gets the table type this provider handles.
+   */
+  String getTableType();
+
+  /**
+   * Creates a table.
+   */
+  void createTable(Table table);
+
+  /**
+   * List all tables from this provider.
+   */
+  List<Table> listTables();
+
+  /**
+   * Build a {@link BeamSqlTable} using the given table meta info.
+   */
+  BeamSqlTable buildBeamSqlTable(Table table);
+
+  /**
+   * Close the provider.
+   */
+  void close();
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaCSVTable.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaCSVTable.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaCSVTable.java
new file mode 100644
index 0000000..a8c8a30
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaCSVTable.java
@@ -0,0 +1,111 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.provider.kafka;
+
+import static org.apache.beam.sdk.extensions.sql.impl.schema.BeamTableUtils.beamRecord2CsvLine;
+import static org.apache.beam.sdk.extensions.sql.impl.schema.BeamTableUtils.csvLine2BeamRecord;
+
+import java.util.List;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.BeamRecord;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.commons.csv.CSVFormat;
+
+/**
+ * A Kafka topic that saves records as CSV format.
+ *
+ */
+public class BeamKafkaCSVTable extends BeamKafkaTable {
+  private CSVFormat csvFormat;
+  public BeamKafkaCSVTable(BeamRecordSqlType beamSqlRowType, String bootstrapServers,
+      List<String> topics) {
+    this(beamSqlRowType, bootstrapServers, topics, CSVFormat.DEFAULT);
+  }
+
+  public BeamKafkaCSVTable(BeamRecordSqlType beamSqlRowType, String bootstrapServers,
+      List<String> topics, CSVFormat format) {
+    super(beamSqlRowType, bootstrapServers, topics);
+    this.csvFormat = format;
+  }
+
+  @Override
+  public PTransform<PCollection<KV<byte[], byte[]>>, PCollection<BeamRecord>>
+      getPTransformForInput() {
+    return new CsvRecorderDecoder(beamRecordSqlType, csvFormat);
+  }
+
+  @Override
+  public PTransform<PCollection<BeamRecord>, PCollection<KV<byte[], byte[]>>>
+      getPTransformForOutput() {
+    return new CsvRecorderEncoder(beamRecordSqlType, csvFormat);
+  }
+
+  /**
+   * A PTransform to convert {@code KV<byte[], byte[]>} to {@link BeamRecord}.
+   *
+   */
+  public static class CsvRecorderDecoder
+      extends PTransform<PCollection<KV<byte[], byte[]>>, PCollection<BeamRecord>> {
+    private BeamRecordSqlType rowType;
+    private CSVFormat format;
+    public CsvRecorderDecoder(BeamRecordSqlType rowType, CSVFormat format) {
+      this.rowType = rowType;
+      this.format = format;
+    }
+
+    @Override
+    public PCollection<BeamRecord> expand(PCollection<KV<byte[], byte[]>> input) {
+      return input.apply("decodeRecord", ParDo.of(new DoFn<KV<byte[], byte[]>, BeamRecord>() {
+        @ProcessElement
+        public void processElement(ProcessContext c) {
+          String rowInString = new String(c.element().getValue());
+          c.output(csvLine2BeamRecord(format, rowInString, rowType));
+        }
+      }));
+    }
+  }
+
+  /**
+   * A PTransform to convert {@link BeamRecord} to {@code KV<byte[], byte[]>}.
+   *
+   */
+  public static class CsvRecorderEncoder
+      extends PTransform<PCollection<BeamRecord>, PCollection<KV<byte[], byte[]>>> {
+    private BeamRecordSqlType rowType;
+    private CSVFormat format;
+    public CsvRecorderEncoder(BeamRecordSqlType rowType, CSVFormat format) {
+      this.rowType = rowType;
+      this.format = format;
+    }
+
+    @Override
+    public PCollection<KV<byte[], byte[]>> expand(PCollection<BeamRecord> input) {
+      return input.apply("encodeRecord", ParDo.of(new DoFn<BeamRecord, KV<byte[], byte[]>>() {
+        @ProcessElement
+        public void processElement(ProcessContext c) {
+          BeamRecord in = c.element();
+          c.output(KV.of(new byte[] {}, beamRecord2CsvLine(in, format).getBytes()));
+        }
+      }));
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaTable.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaTable.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaTable.java
new file mode 100644
index 0000000..50f7496
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaTable.java
@@ -0,0 +1,115 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.provider.kafka;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.extensions.sql.impl.schema.BaseBeamTable;
+import org.apache.beam.sdk.extensions.sql.impl.schema.BeamIOType;
+import org.apache.beam.sdk.io.kafka.KafkaIO;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.BeamRecord;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+
+/**
+ * {@code BeamKafkaTable} represent a Kafka topic, as source or target. Need to
+ * extend to convert between {@code BeamSqlRow} and {@code KV<byte[], byte[]>}.
+ *
+ */
+public abstract class BeamKafkaTable extends BaseBeamTable implements Serializable {
+  private String bootstrapServers;
+  private List<String> topics;
+  private Map<String, Object> configUpdates;
+
+  protected BeamKafkaTable(BeamRecordSqlType beamSqlRowType) {
+    super(beamSqlRowType);
+  }
+
+  public BeamKafkaTable(BeamRecordSqlType beamSqlRowType, String bootstrapServers,
+      List<String> topics) {
+    super(beamSqlRowType);
+    this.bootstrapServers = bootstrapServers;
+    this.topics = topics;
+  }
+
+  public BeamKafkaTable updateConsumerProperties(Map<String, Object> configUpdates) {
+    this.configUpdates = configUpdates;
+    return this;
+  }
+
+  @Override
+  public BeamIOType getSourceType() {
+    return BeamIOType.UNBOUNDED;
+  }
+
+  public abstract PTransform<PCollection<KV<byte[], byte[]>>, PCollection<BeamRecord>>
+      getPTransformForInput();
+
+  public abstract PTransform<PCollection<BeamRecord>, PCollection<KV<byte[], byte[]>>>
+      getPTransformForOutput();
+
+  @Override
+  public PCollection<BeamRecord> buildIOReader(Pipeline pipeline) {
+    return PBegin.in(pipeline).apply("read",
+            KafkaIO.<byte[], byte[]>read()
+                .withBootstrapServers(bootstrapServers)
+                .withTopics(topics)
+                .updateConsumerProperties(configUpdates)
+                .withKeyDeserializerAndCoder(ByteArrayDeserializer.class, ByteArrayCoder.of())
+                .withValueDeserializerAndCoder(ByteArrayDeserializer.class, ByteArrayCoder.of())
+                .withoutMetadata())
+            .apply("in_format", getPTransformForInput());
+  }
+
+  @Override
+  public PTransform<? super PCollection<BeamRecord>, PDone> buildIOWriter() {
+    checkArgument(topics != null && topics.size() == 1,
+        "Only one topic can be acceptable as output.");
+
+    return new PTransform<PCollection<BeamRecord>, PDone>() {
+      @Override
+      public PDone expand(PCollection<BeamRecord> input) {
+        return input.apply("out_reformat", getPTransformForOutput()).apply("persistent",
+            KafkaIO.<byte[], byte[]>write()
+                .withBootstrapServers(bootstrapServers)
+                .withTopic(topics.get(0))
+                .withKeySerializer(ByteArraySerializer.class)
+                .withValueSerializer(ByteArraySerializer.class));
+      }
+    };
+  }
+
+  public String getBootstrapServers() {
+    return bootstrapServers;
+  }
+
+  public List<String> getTopics() {
+    return topics;
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTableProvider.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTableProvider.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTableProvider.java
new file mode 100644
index 0000000..8c37d46
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTableProvider.java
@@ -0,0 +1,82 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.provider.kafka;
+
+import static org.apache.beam.sdk.extensions.sql.meta.provider.MetaUtils.getBeamSqlRecordTypeFromTable;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.extensions.sql.BeamSqlTable;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider;
+
+/**
+ * Kafka table provider.
+ *
+ * <p>A sample of text table is:
+ *
+ * <pre>{@code
+ * CREATE TABLE ORDERS(
+ *   ID INT PRIMARY KEY COMMENT 'this is the primary key',
+ *   NAME VARCHAR(127) COMMENT 'this is the name'
+ * )
+ * COMMENT 'this is the table orders'
+ * LOCATION 'kafka://localhost:2181/brokers?topic=test'
+ * TBLPROPERTIES '{"bootstrap.servers":"localhost:9092", "topics": ["topic1", "topic2"]}'
+ * }</pre>
+ */
+public class KafkaTableProvider implements TableProvider {
+  @Override public BeamSqlTable buildBeamSqlTable(Table table) {
+    BeamRecordSqlType recordType = getBeamSqlRecordTypeFromTable(table);
+
+    JSONObject properties = table.getProperties();
+    String bootstrapServers = properties.getString("bootstrap.servers");
+    JSONArray topicsArr = properties.getJSONArray("topics");
+    List<String> topics = new ArrayList<>(topicsArr.size());
+    for (Object topic : topicsArr) {
+      topics.add(topic.toString());
+    }
+    BeamKafkaCSVTable txtTable = new BeamKafkaCSVTable(recordType, bootstrapServers, topics);
+    return txtTable;
+  }
+
+  @Override public String getTableType() {
+    return "kafka";
+  }
+
+  @Override public void createTable(Table table) {
+    // empty
+  }
+
+  @Override public List<Table> listTables() {
+    return Collections.emptyList();
+  }
+
+  @Override public void init() {
+    // empty
+  }
+
+  @Override public void close() {
+    // empty
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/package-info.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/package-info.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/package-info.java
new file mode 100644
index 0000000..4101da7
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * table schema for KafkaIO.
+ */
+package org.apache.beam.sdk.extensions.sql.meta.provider.kafka;

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/package-info.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/package-info.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/package-info.java
new file mode 100644
index 0000000..c271261
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * Table providers.
+ */
+package org.apache.beam.sdk.extensions.sql.meta.provider;

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextCSVTable.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextCSVTable.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextCSVTable.java
new file mode 100644
index 0000000..78cec74
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextCSVTable.java
@@ -0,0 +1,80 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.provider.text;
+
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.io.TextIO;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.BeamRecord;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
+import org.apache.commons.csv.CSVFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@code BeamTextCSVTable} is a {@code BeamTextTable} which formatted in CSV.
+ *
+ * <p>
+ * {@link CSVFormat} itself has many dialects, check its javadoc for more info.
+ * </p>
+ */
+public class BeamTextCSVTable extends BeamTextTable {
+  private static final Logger LOG = LoggerFactory
+      .getLogger(BeamTextCSVTable.class);
+
+  private String filePattern;
+  private CSVFormat csvFormat;
+
+  /**
+   * CSV table with {@link CSVFormat#DEFAULT DEFAULT} format.
+   */
+  public BeamTextCSVTable(BeamRecordSqlType beamSqlRowType, String filePattern)  {
+    this(beamSqlRowType, filePattern, CSVFormat.DEFAULT);
+  }
+
+  public BeamTextCSVTable(BeamRecordSqlType beamRecordSqlType, String filePattern,
+      CSVFormat csvFormat) {
+    super(beamRecordSqlType, filePattern);
+    this.filePattern = filePattern;
+    this.csvFormat = csvFormat;
+  }
+
+  @Override
+  public PCollection<BeamRecord> buildIOReader(Pipeline pipeline) {
+    return PBegin.in(pipeline).apply("decodeRecord", TextIO.read().from(filePattern))
+        .apply("parseCSVLine",
+            new BeamTextCSVTableIOReader(beamRecordSqlType, filePattern, csvFormat));
+  }
+
+  @Override
+  public PTransform<? super PCollection<BeamRecord>, PDone> buildIOWriter() {
+    return new BeamTextCSVTableIOWriter(beamRecordSqlType, filePattern, csvFormat);
+  }
+
+  public CSVFormat getCsvFormat() {
+    return csvFormat;
+  }
+
+  public String getFilePattern() {
+    return filePattern;
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextCSVTableIOReader.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextCSVTableIOReader.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextCSVTableIOReader.java
new file mode 100644
index 0000000..953ac03
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextCSVTableIOReader.java
@@ -0,0 +1,59 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.provider.text;
+
+import static org.apache.beam.sdk.extensions.sql.impl.schema.BeamTableUtils.csvLine2BeamRecord;
+
+import java.io.Serializable;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.BeamRecord;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.commons.csv.CSVFormat;
+
+/**
+ * IOReader for {@code BeamTextCSVTable}.
+ */
+public class BeamTextCSVTableIOReader
+    extends PTransform<PCollection<String>, PCollection<BeamRecord>>
+    implements Serializable {
+  private String filePattern;
+  protected BeamRecordSqlType beamRecordSqlType;
+  protected CSVFormat csvFormat;
+
+  public BeamTextCSVTableIOReader(BeamRecordSqlType beamRecordSqlType, String filePattern,
+      CSVFormat csvFormat) {
+    this.filePattern = filePattern;
+    this.beamRecordSqlType = beamRecordSqlType;
+    this.csvFormat = csvFormat;
+  }
+
+  @Override
+  public PCollection<BeamRecord> expand(PCollection<String> input) {
+    return input.apply(ParDo.of(new DoFn<String, BeamRecord>() {
+          @ProcessElement
+          public void processElement(ProcessContext ctx) {
+            String str = ctx.element();
+            ctx.output(csvLine2BeamRecord(csvFormat, str, beamRecordSqlType));
+          }
+        }));
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextCSVTableIOWriter.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextCSVTableIOWriter.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextCSVTableIOWriter.java
new file mode 100644
index 0000000..80481d2
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextCSVTableIOWriter.java
@@ -0,0 +1,59 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.provider.text;
+
+import static org.apache.beam.sdk.extensions.sql.impl.schema.BeamTableUtils.beamRecord2CsvLine;
+
+import java.io.Serializable;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.io.TextIO;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.BeamRecord;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
+import org.apache.commons.csv.CSVFormat;
+
+/**
+ * IOWriter for {@code BeamTextCSVTable}.
+ */
+public class BeamTextCSVTableIOWriter extends PTransform<PCollection<BeamRecord>, PDone>
+    implements Serializable {
+  private String filePattern;
+  protected BeamRecordSqlType beamRecordSqlType;
+  protected CSVFormat csvFormat;
+
+  public BeamTextCSVTableIOWriter(BeamRecordSqlType beamRecordSqlType, String filePattern,
+      CSVFormat csvFormat) {
+    this.filePattern = filePattern;
+    this.beamRecordSqlType = beamRecordSqlType;
+    this.csvFormat = csvFormat;
+  }
+
+  @Override public PDone expand(PCollection<BeamRecord> input) {
+    return input.apply("encodeRecord", ParDo.of(new DoFn<BeamRecord, String>() {
+
+      @ProcessElement public void processElement(ProcessContext ctx) {
+        BeamRecord row = ctx.element();
+        ctx.output(beamRecord2CsvLine(row, csvFormat));
+      }
+    })).apply(TextIO.write().to(filePattern));
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextTable.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextTable.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextTable.java
new file mode 100644
index 0000000..76616ef
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/BeamTextTable.java
@@ -0,0 +1,41 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.provider.text;
+
+import java.io.Serializable;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.extensions.sql.impl.schema.BaseBeamTable;
+import org.apache.beam.sdk.extensions.sql.impl.schema.BeamIOType;
+
+/**
+ * {@code BeamTextTable} represents a text file/directory(backed by {@code TextIO}).
+ */
+public abstract class BeamTextTable extends BaseBeamTable implements Serializable {
+  protected String filePattern;
+
+  protected BeamTextTable(BeamRecordSqlType beamRecordSqlType, String filePattern) {
+    super(beamRecordSqlType);
+    this.filePattern = filePattern;
+  }
+
+  @Override
+  public BeamIOType getSourceType() {
+    return BeamIOType.BOUNDED;
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/TextTableProvider.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/TextTableProvider.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/TextTableProvider.java
new file mode 100644
index 0000000..bc9f03f
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/TextTableProvider.java
@@ -0,0 +1,83 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.provider.text;
+
+import static org.apache.beam.sdk.extensions.sql.meta.provider.MetaUtils.getBeamSqlRecordTypeFromTable;
+
+import com.alibaba.fastjson.JSONObject;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.extensions.sql.BeamSqlTable;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider;
+import org.apache.commons.csv.CSVFormat;
+
+/**
+ * Text table provider.
+ *
+ * <p>A sample of text table is:
+ * <pre>{@code
+ * CREATE TABLE ORDERS(
+ *   ID INT PRIMARY KEY COMMENT 'this is the primary key',
+ *   NAME VARCHAR(127) COMMENT 'this is the name'
+ * )
+ * TYPE 'text'
+ * COMMENT 'this is the table orders'
+ * LOCATION 'text://home/admin/orders'
+ * TBLPROPERTIES '{"format": "Excel"}' -- format of each text line(csv format)
+ * }</pre>
+ */
+public class TextTableProvider implements TableProvider {
+
+  @Override public String getTableType() {
+    return "text";
+  }
+
+  @Override public BeamSqlTable buildBeamSqlTable(Table table) {
+    BeamRecordSqlType recordType = getBeamSqlRecordTypeFromTable(table);
+
+    String filePattern = table.getLocationAsString();
+    CSVFormat format = CSVFormat.DEFAULT;
+    JSONObject properties = table.getProperties();
+    String csvFormatStr = properties.getString("format");
+    if (csvFormatStr != null && !csvFormatStr.isEmpty()) {
+      format = CSVFormat.valueOf(csvFormatStr);
+    }
+
+    BeamTextCSVTable txtTable = new BeamTextCSVTable(recordType, filePattern, format);
+    return txtTable;
+  }
+
+  @Override public void createTable(Table table) {
+    // empty
+  }
+
+  @Override public List<Table> listTables() {
+    return Collections.emptyList();
+  }
+
+  @Override public void init() {
+    // empty
+  }
+
+  @Override public void close() {
+    // empty
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/package-info.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/package-info.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/package-info.java
new file mode 100644
index 0000000..2dd9e6e
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * Table schema for text files.
+ */
+package org.apache.beam.sdk.extensions.sql.meta.provider.text;

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/store/InMemoryMetaStore.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/store/InMemoryMetaStore.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/store/InMemoryMetaStore.java
new file mode 100644
index 0000000..bacfbff
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/store/InMemoryMetaStore.java
@@ -0,0 +1,113 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.store;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.beam.sdk.extensions.sql.BeamSqlTable;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider;
+
+/**
+ * A {@link MetaStore} which stores the meta info in memory.
+ *
+ * <p>NOTE, because this implementation is memory based, the metadata is NOT persistent.
+ * for tables which created, you need to create again every time you launch the
+ * {@link org.apache.beam.sdk.extensions.sql.BeamSqlCli}.
+ */
+public class InMemoryMetaStore implements MetaStore {
+  private Map<String, Table> tables = new HashMap<>();
+  private Map<String, TableProvider> providers = new HashMap<>();
+
+  public InMemoryMetaStore() {
+  }
+
+  @Override public void createTable(Table table) {
+    validateTableType(table);
+
+    // first assert the table name is unique
+    if (tables.containsKey(table.getName())) {
+      throw new IllegalArgumentException("Duplicate table name: " + table.getName());
+    }
+
+    // invoke the provider's create
+    providers.get(table.getType()).createTable(table);
+
+    // store to the global metastore
+    tables.put(table.getName(), table);
+  }
+
+  @Override public Table getTable(String tableName) {
+    if (tableName == null) {
+      return null;
+    }
+    return tables.get(tableName.toLowerCase());
+  }
+
+  @Override public List<Table> listTables() {
+    return new ArrayList<>(tables.values());
+  }
+
+  @Override public BeamSqlTable buildBeamSqlTable(String tableName) {
+    Table table = getTable(tableName);
+
+    if (table == null) {
+      throw new IllegalArgumentException("The specified table: " + tableName + " does not exists!");
+    }
+
+    TableProvider provider = providers.get(table.getType());
+
+    return provider.buildBeamSqlTable(table);
+  }
+
+  private void validateTableType(Table table) {
+    if (!providers.containsKey(table.getType())) {
+      throw new IllegalArgumentException(
+          "Table type: " + table.getType() + " not supported!");
+    }
+  }
+
+  public void registerProvider(TableProvider provider) {
+    if (providers.containsKey(provider.getTableType())) {
+      throw new IllegalArgumentException("Provider is already registered for table type: "
+          + provider.getTableType());
+    }
+
+    this.providers.put(provider.getTableType(), provider);
+    initTablesFromProvider(provider);
+  }
+
+  private void initTablesFromProvider(TableProvider provider) {
+    List<Table> tables = provider.listTables();
+    for (Table table : tables) {
+      if (this.tables.containsKey(table.getName())) {
+        throw new IllegalStateException(
+            "Duplicate table: " + table.getName() + " from provider: " + provider);
+      }
+
+      this.tables.put(table.getName(), table);
+    }
+  }
+
+  Map<String, TableProvider> getProviders() {
+    return providers;
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/store/MetaStore.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/store/MetaStore.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/store/MetaStore.java
new file mode 100644
index 0000000..2f395f0
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/store/MetaStore.java
@@ -0,0 +1,56 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.store;
+
+import java.util.List;
+import org.apache.beam.sdk.extensions.sql.BeamSqlTable;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider;
+
+/**
+ * The interface to handle CRUD of {@code BeamSql} table metadata.
+ */
+public interface MetaStore {
+
+  /**
+   * create a table.
+   */
+  void createTable(Table table);
+
+  /**
+   * Get table with the specified name.
+   */
+  Table getTable(String tableName);
+
+  /**
+   * List all the tables.
+   */
+  List<Table> listTables();
+
+  /**
+   * Build the {@code BeamSqlTable} for the specified table.
+   */
+  BeamSqlTable buildBeamSqlTable(String tableName);
+
+  /**
+   * Register a table provider.
+   * @param provider
+   */
+  void registerProvider(TableProvider provider);
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/store/package-info.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/store/package-info.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/store/package-info.java
new file mode 100644
index 0000000..39f1385
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/store/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * Meta stores.
+ */
+package org.apache.beam.sdk.extensions.sql.meta.store;

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlApiSurfaceTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlApiSurfaceTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlApiSurfaceTest.java
index 456662f..156d7ff 100644
--- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlApiSurfaceTest.java
+++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlApiSurfaceTest.java
@@ -40,10 +40,15 @@ public class BeamSqlApiSurfaceTest {
     final Set<String> allowed =
         ImmutableSet.of(
             "org.apache.beam",
-            "org.joda.time");
+            "org.joda.time",
+            "com.alibaba.fastjson",
+            // exposed by fastjson
+            "sun.reflect"
+            );
 
     ApiSurface surface = ApiSurface
         .ofClass(BeamSql.class)
+        .includingClass(BeamSqlCli.class)
         .includingClass(BeamSqlUdf.class)
         .includingClass(BeamRecordSqlType.class)
         .includingClass(BeamSqlRecordHelper.class)

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlCliTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlCliTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlCliTest.java
new file mode 100644
index 0000000..62d6933
--- /dev/null
+++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlCliTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.beam.sdk.extensions.sql;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+import org.apache.beam.sdk.extensions.sql.meta.provider.text.TextTableProvider;
+import org.apache.beam.sdk.extensions.sql.meta.store.InMemoryMetaStore;
+import org.junit.Test;
+
+/**
+ * UnitTest for {@link BeamSqlCli}.
+ */
+public class BeamSqlCliTest {
+  @Test
+  public void testExecute_createTextTable() throws Exception {
+    InMemoryMetaStore metaStore = new InMemoryMetaStore();
+    metaStore.registerProvider(new TextTableProvider());
+
+    BeamSqlCli cli = new BeamSqlCli()
+        .metaStore(metaStore);
+    cli.execute(
+        "create table person (\n"
+        + "id int COMMENT 'id', \n"
+        + "name varchar(31) COMMENT 'name', \n"
+        + "age int COMMENT 'age') \n"
+        + "TYPE 'text' \n"
+        + "COMMENT '' LOCATION 'text://home/admin/orders'"
+    );
+    Table table = metaStore.getTable("person");
+    assertNotNull(table);
+  }
+
+  @Test
+  public void testExplainQuery() throws Exception {
+    InMemoryMetaStore metaStore = new InMemoryMetaStore();
+    metaStore.registerProvider(new TextTableProvider());
+
+    BeamSqlCli cli = new BeamSqlCli()
+        .metaStore(metaStore);
+
+    cli.execute(
+        "create table person (\n"
+            + "id int COMMENT 'id', \n"
+            + "name varchar(31) COMMENT 'name', \n"
+            + "age int COMMENT 'age') \n"
+            + "TYPE 'text' \n"
+            + "COMMENT '' LOCATION 'text://home/admin/orders'"
+    );
+
+    String plan = cli.explainQuery("select * from person");
+    assertEquals(
+        "BeamProjectRel(id=[$0], name=[$1], age=[$2])\n"
+        + "  BeamIOSourceRel(table=[[person]])\n",
+        plan
+    );
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/TestUtils.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/TestUtils.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/TestUtils.java
index 4a1f8a0..d4cc53a 100644
--- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/TestUtils.java
+++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/TestUtils.java
@@ -98,7 +98,7 @@ public class TestUtils {
      * <p>For example:
      * <pre>{@code
      * TestUtils.RowsBuilder.of(
-     *   beamSqlRowType
+     *   beamRecordSqlType
      * )}</pre>
      * @beamSQLRowType the record type.
      */

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/parser/BeamSqlParserTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/parser/BeamSqlParserTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/parser/BeamSqlParserTest.java
new file mode 100644
index 0000000..c7c8bf4
--- /dev/null
+++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/parser/BeamSqlParserTest.java
@@ -0,0 +1,167 @@
+/*
+ * 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.beam.sdk.extensions.sql.impl.parser;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.google.common.collect.ImmutableList;
+import java.net.URI;
+import java.sql.Types;
+import org.apache.beam.sdk.extensions.sql.meta.Column;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+import org.apache.calcite.sql.SqlNode;
+import org.junit.Test;
+
+/**
+ * UnitTest for {@link BeamSqlParser}.
+ */
+public class BeamSqlParserTest {
+  @Test
+  public void testParseCreateTable_full() throws Exception {
+    JSONObject properties = new JSONObject();
+    JSONArray hello = new JSONArray();
+    hello.add("james");
+    hello.add("bond");
+    properties.put("hello", hello);
+
+    Table table = parseTable(
+        "create table person (\n"
+            + "id int COMMENT 'id', \n"
+            + "name varchar(31) COMMENT 'name') \n"
+            + "TYPE 'text' \n"
+            + "COMMENT 'person table' \n"
+            + "LOCATION 'text://home/admin/person'\n"
+            + "TBLPROPERTIES '{\"hello\": [\"james\", \"bond\"]}'"
+    );
+    assertEquals(
+        mockTable("person", "text", "person table", properties),
+        table
+    );
+  }
+
+  @Test(expected = org.apache.beam.sdk.extensions.sql.impl.parser.impl.ParseException.class)
+  public void testParseCreateTable_withoutType() throws Exception {
+    parseTable(
+        "create table person (\n"
+            + "id int COMMENT 'id', \n"
+            + "name varchar(31) COMMENT 'name') \n"
+            + "COMMENT 'person table' \n"
+            + "LOCATION 'text://home/admin/person'\n"
+            + "TBLPROPERTIES '{\"hello\": [\"james\", \"bond\"]}'"
+    );
+  }
+
+  @Test
+  public void testParseCreateTable_withoutTableComment() throws Exception {
+    JSONObject properties = new JSONObject();
+    JSONArray hello = new JSONArray();
+    hello.add("james");
+    hello.add("bond");
+    properties.put("hello", hello);
+
+    Table table = parseTable(
+        "create table person (\n"
+            + "id int COMMENT 'id', \n"
+            + "name varchar(31) COMMENT 'name') \n"
+            + "TYPE 'text' \n"
+            + "LOCATION 'text://home/admin/person'\n"
+            + "TBLPROPERTIES '{\"hello\": [\"james\", \"bond\"]}'"
+    );
+    assertEquals(mockTable("person", "text", null, properties), table);
+  }
+
+  @Test
+  public void testParseCreateTable_withoutTblProperties() throws Exception {
+    Table table = parseTable(
+        "create table person (\n"
+            + "id int COMMENT 'id', \n"
+            + "name varchar(31) COMMENT 'name') \n"
+            + "TYPE 'text' \n"
+            + "COMMENT 'person table' \n"
+            + "LOCATION 'text://home/admin/person'\n"
+    );
+    assertEquals(
+        mockTable("person", "text", "person table", new JSONObject()),
+        table
+    );
+  }
+
+  @Test
+  public void testParseCreateTable_withoutLocation() throws Exception {
+    Table table = parseTable(
+        "create table person (\n"
+            + "id int COMMENT 'id', \n"
+            + "name varchar(31) COMMENT 'name') \n"
+            + "TYPE 'text' \n"
+            + "COMMENT 'person table' \n"
+    );
+
+    assertEquals(
+        mockTable("person", "text", "person table", new JSONObject(), null),
+        table
+    );
+  }
+
+  private Table parseTable(String sql) throws Exception {
+    BeamSqlParser parser = new BeamSqlParser(sql);
+    SqlNode sqlNode = parser.impl().parseSqlStmtEof();
+
+    assertNotNull(sqlNode);
+    assertTrue(sqlNode instanceof SqlCreateTable);
+    SqlCreateTable stmt = (SqlCreateTable) sqlNode;
+    return ParserUtils.convertCreateTableStmtToTable(stmt);
+  }
+
+  private static Table mockTable(String name, String type, String comment, JSONObject properties) {
+    return mockTable(name, type, comment, properties, "text://home/admin/" + name);
+  }
+
+  private static Table mockTable(String name, String type, String comment, JSONObject properties,
+      String location) {
+    URI locationURI = null;
+    if (location != null) {
+      locationURI = URI.create(location);
+    }
+
+    return Table.builder()
+        .name(name)
+        .type(type)
+        .comment(comment)
+        .location(locationURI)
+        .columns(ImmutableList.of(
+            Column.builder()
+                .name("id")
+                .type(Types.INTEGER)
+                .primaryKey(false)
+                .comment("id")
+                .build(),
+            Column.builder()
+                .name("name")
+                .type(Types.VARCHAR)
+                .primaryKey(false)
+                .comment("name")
+                .build()
+        ))
+        .properties(properties)
+        .build();
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/schema/kafka/BeamKafkaCSVTableTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/schema/kafka/BeamKafkaCSVTableTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/schema/kafka/BeamKafkaCSVTableTest.java
deleted file mode 100644
index fd88448..0000000
--- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/schema/kafka/BeamKafkaCSVTableTest.java
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * 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.beam.sdk.extensions.sql.impl.schema.kafka;
-
-import java.io.Serializable;
-import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
-import org.apache.beam.sdk.extensions.sql.impl.planner.BeamQueryPlanner;
-import org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils;
-import org.apache.beam.sdk.testing.PAssert;
-import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.values.BeamRecord;
-import org.apache.beam.sdk.values.KV;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.rel.type.RelDataTypeFactory;
-import org.apache.calcite.rel.type.RelProtoDataType;
-import org.apache.calcite.sql.type.SqlTypeName;
-import org.apache.commons.csv.CSVFormat;
-import org.junit.BeforeClass;
-import org.junit.Rule;
-import org.junit.Test;
-
-/**
- * Test for BeamKafkaCSVTable.
- */
-public class BeamKafkaCSVTableTest {
-  @Rule
-  public TestPipeline pipeline = TestPipeline.create();
-  public static BeamRecord row1;
-  public static BeamRecord row2;
-
-  @BeforeClass
-  public static void setUp() {
-    row1 = new BeamRecord(genRowType(), 1L, 1, 1.0);
-
-    row2 = new BeamRecord(genRowType(), 2L, 2, 2.0);
-  }
-
-  @Test public void testCsvRecorderDecoder() throws Exception {
-    PCollection<BeamRecord> result = pipeline
-        .apply(
-            Create.of("1,\"1\",1.0", "2,2,2.0")
-        )
-        .apply(ParDo.of(new String2KvBytes()))
-        .apply(
-            new BeamKafkaCSVTable.CsvRecorderDecoder(genRowType(), CSVFormat.DEFAULT)
-        );
-
-    PAssert.that(result).containsInAnyOrder(row1, row2);
-
-    pipeline.run();
-  }
-
-  @Test public void testCsvRecorderEncoder() throws Exception {
-    PCollection<BeamRecord> result = pipeline
-        .apply(
-            Create.of(row1, row2)
-        )
-        .apply(
-            new BeamKafkaCSVTable.CsvRecorderEncoder(genRowType(), CSVFormat.DEFAULT)
-        ).apply(
-            new BeamKafkaCSVTable.CsvRecorderDecoder(genRowType(), CSVFormat.DEFAULT)
-        );
-
-    PAssert.that(result).containsInAnyOrder(row1, row2);
-
-    pipeline.run();
-  }
-
-  private static BeamRecordSqlType genRowType() {
-    return CalciteUtils.toBeamRowType(new RelProtoDataType() {
-
-      @Override public RelDataType apply(RelDataTypeFactory a0) {
-        return a0.builder().add("order_id", SqlTypeName.BIGINT)
-            .add("site_id", SqlTypeName.INTEGER)
-            .add("price", SqlTypeName.DOUBLE).build();
-      }
-    }.apply(BeamQueryPlanner.TYPE_FACTORY));
-  }
-
-  private static class String2KvBytes extends DoFn<String, KV<byte[], byte[]>>
-      implements Serializable {
-    @ProcessElement
-    public void processElement(ProcessContext ctx) {
-      ctx.output(KV.of(new byte[] {}, ctx.element().getBytes()));
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/schema/text/BeamTextCSVTableTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/schema/text/BeamTextCSVTableTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/schema/text/BeamTextCSVTableTest.java
deleted file mode 100644
index 9a57a5f..0000000
--- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/schema/text/BeamTextCSVTableTest.java
+++ /dev/null
@@ -1,176 +0,0 @@
-/*
- * 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.beam.sdk.extensions.sql.impl.schema.text;
-
-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.io.PrintStream;
-import java.nio.file.FileVisitResult;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.SimpleFileVisitor;
-import java.nio.file.attribute.BasicFileAttributes;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
-import org.apache.beam.sdk.extensions.sql.impl.planner.BeamQueryPlanner;
-import org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils;
-import org.apache.beam.sdk.testing.PAssert;
-import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.values.BeamRecord;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.rel.type.RelDataTypeFactory;
-import org.apache.calcite.rel.type.RelProtoDataType;
-import org.apache.calcite.sql.type.SqlTypeName;
-import org.apache.commons.csv.CSVFormat;
-import org.apache.commons.csv.CSVPrinter;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Rule;
-import org.junit.Test;
-
-/**
- * Tests for {@code BeamTextCSVTable}.
- */
-public class BeamTextCSVTableTest {
-
-  @Rule public TestPipeline pipeline = TestPipeline.create();
-  @Rule public TestPipeline pipeline2 = TestPipeline.create();
-
-  /**
-   * testData.
-   *
-   * <p>
-   * The types of the csv fields are:
-   *     integer,bigint,float,double,string
-   * </p>
-   */
-  private static Object[] data1 = new Object[] { 1, 1L, 1.1F, 1.1, "james" };
-  private static Object[] data2 = new Object[] { 2, 2L, 2.2F, 2.2, "bond" };
-
-  private static List<Object[]> testData = Arrays.asList(data1, data2);
-  private static List<BeamRecord> testDataRows = new ArrayList<BeamRecord>() {{
-    for (Object[] data : testData) {
-      add(buildRow(data));
-    }
-  }};
-
-  private static Path tempFolder;
-  private static File readerSourceFile;
-  private static File writerTargetFile;
-
-  @Test public void testBuildIOReader() {
-    PCollection<BeamRecord> rows = new BeamTextCSVTable(buildBeamSqlRowType(),
-        readerSourceFile.getAbsolutePath()).buildIOReader(pipeline);
-    PAssert.that(rows).containsInAnyOrder(testDataRows);
-    pipeline.run();
-  }
-
-  @Test public void testBuildIOWriter() {
-    new BeamTextCSVTable(buildBeamSqlRowType(),
-        readerSourceFile.getAbsolutePath()).buildIOReader(pipeline)
-        .apply(new BeamTextCSVTable(buildBeamSqlRowType(), writerTargetFile.getAbsolutePath())
-            .buildIOWriter());
-    pipeline.run();
-
-    PCollection<BeamRecord> rows = new BeamTextCSVTable(buildBeamSqlRowType(),
-        writerTargetFile.getAbsolutePath()).buildIOReader(pipeline2);
-
-    // confirm the two reads match
-    PAssert.that(rows).containsInAnyOrder(testDataRows);
-    pipeline2.run();
-  }
-
-  @BeforeClass public static void setUp() throws IOException {
-    tempFolder = Files.createTempDirectory("BeamTextTableTest");
-    readerSourceFile = writeToFile(testData, "readerSourceFile.txt");
-    writerTargetFile = writeToFile(testData, "writerTargetFile.txt");
-  }
-
-  @AfterClass public static void teardownClass() throws IOException {
-    Files.walkFileTree(tempFolder, new SimpleFileVisitor<Path>() {
-
-      @Override public FileVisitResult visitFile(Path file, BasicFileAttributes attrs)
-          throws IOException {
-        Files.delete(file);
-        return FileVisitResult.CONTINUE;
-      }
-
-      @Override public FileVisitResult postVisitDirectory(Path dir, IOException exc)
-          throws IOException {
-        Files.delete(dir);
-        return FileVisitResult.CONTINUE;
-      }
-    });
-  }
-
-  private static File writeToFile(List<Object[]> rows, String filename) throws IOException {
-    File file = tempFolder.resolve(filename).toFile();
-    OutputStream output = new FileOutputStream(file);
-    writeToStreamAndClose(rows, output);
-    return file;
-  }
-
-  /**
-   * Helper that writes the given lines (adding a newline in between) to a stream, then closes the
-   * stream.
-   */
-  private static void writeToStreamAndClose(List<Object[]> rows, OutputStream outputStream) {
-    try (PrintStream writer = new PrintStream(outputStream)) {
-      CSVPrinter printer = CSVFormat.DEFAULT.print(writer);
-      for (Object[] row : rows) {
-        for (Object field : row) {
-          printer.print(field);
-        }
-        printer.println();
-      }
-    } catch (IOException e) {
-      e.printStackTrace();
-    }
-  }
-
-  private RelProtoDataType buildRowType() {
-    return new RelProtoDataType() {
-
-      @Override public RelDataType apply(RelDataTypeFactory a0) {
-        return a0.builder().add("id", SqlTypeName.INTEGER).add("order_id", SqlTypeName.BIGINT)
-            .add("price", SqlTypeName.FLOAT).add("amount", SqlTypeName.DOUBLE)
-            .add("user_name", SqlTypeName.VARCHAR).build();
-      }
-    };
-  }
-
-  private static RelDataType buildRelDataType() {
-    return BeamQueryPlanner.TYPE_FACTORY.builder().add("id", SqlTypeName.INTEGER)
-        .add("order_id", SqlTypeName.BIGINT).add("price", SqlTypeName.FLOAT)
-        .add("amount", SqlTypeName.DOUBLE).add("user_name", SqlTypeName.VARCHAR).build();
-  }
-
-  private static BeamRecordSqlType buildBeamSqlRowType() {
-    return CalciteUtils.toBeamRowType(buildRelDataType());
-  }
-
-  private static BeamRecord buildRow(Object[] data) {
-    return new BeamRecord(buildBeamSqlRowType(), Arrays.asList(data));
-  }
-}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaCSVTableTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaCSVTableTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaCSVTableTest.java
new file mode 100644
index 0000000..8ce4364
--- /dev/null
+++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaCSVTableTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.provider.kafka;
+
+import java.io.Serializable;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.extensions.sql.impl.planner.BeamQueryPlanner;
+import org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.BeamRecord;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelProtoDataType;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.commons.csv.CSVFormat;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+
+/**
+ * Test for BeamKafkaCSVTable.
+ */
+public class BeamKafkaCSVTableTest {
+  @Rule
+  public TestPipeline pipeline = TestPipeline.create();
+  public static BeamRecord row1;
+  public static BeamRecord row2;
+
+  @BeforeClass
+  public static void setUp() {
+    row1 = new BeamRecord(genRowType(), 1L, 1, 1.0);
+
+    row2 = new BeamRecord(genRowType(), 2L, 2, 2.0);
+  }
+
+  @Test public void testCsvRecorderDecoder() throws Exception {
+    PCollection<BeamRecord> result = pipeline
+        .apply(
+            Create.of("1,\"1\",1.0", "2,2,2.0")
+        )
+        .apply(ParDo.of(new String2KvBytes()))
+        .apply(
+            new BeamKafkaCSVTable.CsvRecorderDecoder(genRowType(), CSVFormat.DEFAULT)
+        );
+
+    PAssert.that(result).containsInAnyOrder(row1, row2);
+
+    pipeline.run();
+  }
+
+  @Test public void testCsvRecorderEncoder() throws Exception {
+    PCollection<BeamRecord> result = pipeline
+        .apply(
+            Create.of(row1, row2)
+        )
+        .apply(
+            new BeamKafkaCSVTable.CsvRecorderEncoder(genRowType(), CSVFormat.DEFAULT)
+        ).apply(
+            new BeamKafkaCSVTable.CsvRecorderDecoder(genRowType(), CSVFormat.DEFAULT)
+        );
+
+    PAssert.that(result).containsInAnyOrder(row1, row2);
+
+    pipeline.run();
+  }
+
+  private static BeamRecordSqlType genRowType() {
+    return CalciteUtils.toBeamRowType(new RelProtoDataType() {
+
+      @Override public RelDataType apply(RelDataTypeFactory a0) {
+        return a0.builder().add("order_id", SqlTypeName.BIGINT)
+            .add("site_id", SqlTypeName.INTEGER)
+            .add("price", SqlTypeName.DOUBLE).build();
+      }
+    }.apply(BeamQueryPlanner.TYPE_FACTORY));
+  }
+
+  private static class String2KvBytes extends DoFn<String, KV<byte[], byte[]>>
+      implements Serializable {
+    @ProcessElement
+    public void processElement(ProcessContext ctx) {
+      ctx.output(KV.of(new byte[] {}, ctx.element().getBytes()));
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/208fc38a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTableProviderTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTableProviderTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTableProviderTest.java
new file mode 100644
index 0000000..a7c2719
--- /dev/null
+++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTableProviderTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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.beam.sdk.extensions.sql.meta.provider.kafka;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.google.common.collect.ImmutableList;
+import java.net.URI;
+import java.sql.Types;
+import org.apache.beam.sdk.extensions.sql.BeamSqlTable;
+import org.apache.beam.sdk.extensions.sql.meta.Column;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+import org.junit.Test;
+
+/**
+ * UnitTest for {@link KafkaTableProvider}.
+ */
+public class KafkaTableProviderTest {
+  private KafkaTableProvider provider = new KafkaTableProvider();
+  @Test public void testBuildBeamSqlTable() throws Exception {
+    Table table = mockTable("hello");
+    BeamSqlTable sqlTable = provider.buildBeamSqlTable(table);
+
+    assertNotNull(sqlTable);
+    assertTrue(sqlTable instanceof BeamKafkaCSVTable);
+
+    BeamKafkaCSVTable csvTable = (BeamKafkaCSVTable) sqlTable;
+    assertEquals("localhost:9092", csvTable.getBootstrapServers());
+    assertEquals(ImmutableList.of("topic1", "topic2"), csvTable.getTopics());
+  }
+
+  @Test
+  public void testGetTableType() throws Exception {
+    assertEquals("kafka", provider.getTableType());
+  }
+
+  private static Table mockTable(String name) {
+    JSONObject properties = new JSONObject();
+    properties.put("bootstrap.servers", "localhost:9092");
+    JSONArray topics = new JSONArray();
+    topics.add("topic1");
+    topics.add("topic2");
+    properties.put("topics", topics);
+
+    return Table.builder()
+        .name(name)
+        .comment(name + " table")
+        .location(URI.create("kafka://localhost:2181/brokers?topic=test"))
+        .columns(ImmutableList.of(
+            Column.builder().name("id").type(Types.INTEGER).primaryKey(true).build(),
+            Column.builder().name("name").type(Types.VARCHAR).primaryKey(false).build()
+        ))
+        .type("kafka")
+        .properties(properties)
+        .build();
+  }
+}