beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ming...@apache.org
Subject [58/66] [abbrv] beam git commit: [BEAM-2740] Hide BeamSqlEnv.
Date Mon, 11 Sep 2017 20:19:41 GMT
[BEAM-2740] Hide BeamSqlEnv.


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/b17cf303
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/b17cf303
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/b17cf303

Branch: refs/heads/master
Commit: b17cf303bcf458fc14d8d6c529ba18b0418a52e5
Parents: bbf205d
Author: James Xu <xumingmingv@gmail.com>
Authored: Wed Aug 9 18:54:54 2017 +0800
Committer: mingmxu <mingmxu@ebay.com>
Committed: Mon Sep 11 10:56:58 2017 -0700

----------------------------------------------------------------------
 .../sdk/extensions/sql/BeamRecordSqlType.java   | 185 ++++++++
 .../apache/beam/sdk/extensions/sql/BeamSql.java | 113 ++---
 .../beam/sdk/extensions/sql/BeamSqlCli.java     |  65 ---
 .../beam/sdk/extensions/sql/BeamSqlEnv.java     | 127 ------
 .../sdk/extensions/sql/BeamSqlRecordHelper.java | 217 +++++++++
 .../beam/sdk/extensions/sql/BeamSqlUdf.java     |  41 ++
 .../extensions/sql/example/BeamSqlExample.java  |   2 +-
 .../sdk/extensions/sql/impl/BeamSqlCli.java     |  65 +++
 .../sdk/extensions/sql/impl/BeamSqlEnv.java     | 135 ++++++
 .../sdk/extensions/sql/impl/package-info.java   |  22 +
 .../sql/impl/planner/BeamQueryPlanner.java      |   9 +-
 .../sql/impl/rel/BeamAggregationRel.java        |   4 +-
 .../extensions/sql/impl/rel/BeamFilterRel.java  |   2 +-
 .../extensions/sql/impl/rel/BeamIOSinkRel.java  |   6 +-
 .../sql/impl/rel/BeamIOSourceRel.java           |   6 +-
 .../sql/impl/rel/BeamIntersectRel.java          |   2 +-
 .../extensions/sql/impl/rel/BeamJoinRel.java    |   4 +-
 .../extensions/sql/impl/rel/BeamMinusRel.java   |   2 +-
 .../extensions/sql/impl/rel/BeamProjectRel.java |   2 +-
 .../extensions/sql/impl/rel/BeamRelNode.java    |   5 +-
 .../sql/impl/rel/BeamSetOperatorRelBase.java    |   2 +-
 .../extensions/sql/impl/rel/BeamSortRel.java    |   4 +-
 .../extensions/sql/impl/rel/BeamUnionRel.java   |   2 +-
 .../extensions/sql/impl/rel/BeamValuesRel.java  |   6 +-
 .../sql/impl/schema/BaseBeamTable.java          |  35 ++
 .../extensions/sql/impl/schema/BeamIOType.java  |  28 ++
 .../sql/impl/schema/BeamPCollectionTable.java   |  63 +++
 .../sql/impl/schema/BeamSqlTable.java           |  54 +++
 .../sql/impl/schema/BeamTableUtils.java         | 118 +++++
 .../impl/schema/kafka/BeamKafkaCSVTable.java    | 109 +++++
 .../sql/impl/schema/kafka/BeamKafkaTable.java   | 109 +++++
 .../sql/impl/schema/kafka/package-info.java     |  22 +
 .../sql/impl/schema/package-info.java           |  22 +
 .../sql/impl/schema/text/BeamTextCSVTable.java  |  70 +++
 .../schema/text/BeamTextCSVTableIOReader.java   |  58 +++
 .../schema/text/BeamTextCSVTableIOWriter.java   |  58 +++
 .../sql/impl/schema/text/BeamTextTable.java     |  41 ++
 .../sql/impl/schema/text/package-info.java      |  22 +
 .../transform/BeamAggregationTransforms.java    |   4 +-
 .../sql/impl/transform/BeamJoinTransforms.java  |   4 +-
 .../sql/impl/transform/BeamSqlProjectFn.java    |   4 +-
 .../extensions/sql/impl/utils/CalciteUtils.java |   2 +-
 .../extensions/sql/schema/BaseBeamTable.java    |  34 --
 .../sdk/extensions/sql/schema/BeamIOType.java   |  28 --
 .../sql/schema/BeamPCollectionTable.java        |  62 ---
 .../sql/schema/BeamRecordSqlType.java           | 185 --------
 .../sql/schema/BeamSqlRecordHelper.java         | 217 ---------
 .../sdk/extensions/sql/schema/BeamSqlTable.java |  53 ---
 .../sdk/extensions/sql/schema/BeamSqlUdf.java   |  41 --
 .../extensions/sql/schema/BeamTableUtils.java   | 117 -----
 .../sql/schema/kafka/BeamKafkaCSVTable.java     | 109 -----
 .../sql/schema/kafka/BeamKafkaTable.java        | 109 -----
 .../sql/schema/kafka/package-info.java          |  22 -
 .../sdk/extensions/sql/schema/package-info.java |  22 -
 .../sql/schema/text/BeamTextCSVTable.java       |  70 ---
 .../schema/text/BeamTextCSVTableIOReader.java   |  58 ---
 .../schema/text/BeamTextCSVTableIOWriter.java   |  58 ---
 .../sql/schema/text/BeamTextTable.java          |  41 --
 .../sql/schema/text/package-info.java           |  22 -
 .../extensions/sql/BeamSqlApiSurfaceTest.java   |  12 +-
 .../sql/BeamSqlDslAggregationTest.java          |   1 -
 .../beam/sdk/extensions/sql/BeamSqlDslBase.java |   1 -
 .../sdk/extensions/sql/BeamSqlDslJoinTest.java  |   1 -
 .../extensions/sql/BeamSqlDslProjectTest.java   |   1 -
 .../extensions/sql/BeamSqlDslUdfUdafTest.java   |   2 -
 .../beam/sdk/extensions/sql/TestUtils.java      |   1 -
 .../interpreter/BeamSqlFnExecutorTestBase.java  |   2 +-
 .../extensions/sql/impl/rel/BaseRelTest.java    |  34 ++
 .../sql/impl/rel/BeamIntersectRelTest.java      |   9 +-
 .../rel/BeamJoinRelBoundedVsBoundedTest.java    |  23 +-
 .../rel/BeamJoinRelUnboundedVsBoundedTest.java  |  25 +-
 .../BeamJoinRelUnboundedVsUnboundedTest.java    |  19 +-
 .../sql/impl/rel/BeamMinusRelTest.java          |   9 +-
 .../impl/rel/BeamSetOperatorRelBaseTest.java    |   9 +-
 .../sql/impl/rel/BeamSortRelTest.java           |  17 +-
 .../sql/impl/rel/BeamUnionRelTest.java          |   9 +-
 .../sql/impl/rel/BeamValuesRelTest.java         |  11 +-
 .../sql/impl/schema/BeamSqlRowCoderTest.java    |  77 ++++
 .../schema/kafka/BeamKafkaCSVTableTest.java     | 107 +++++
 .../impl/schema/text/BeamTextCSVTableTest.java  | 176 +++++++
 .../transform/BeamAggregationTransformTest.java | 453 +++++++++++++++++++
 .../schema/transform/BeamTransformBaseTest.java |  97 ++++
 ...mSqlBuiltinFunctionsIntegrationTestBase.java |   2 +-
 ...amSqlComparisonOperatorsIntegrationTest.java |   2 +-
 .../extensions/sql/mock/MockedBoundedTable.java |   4 +-
 .../sdk/extensions/sql/mock/MockedTable.java    |   4 +-
 .../sql/mock/MockedUnboundedTable.java          |   4 +-
 .../sql/schema/BeamSqlRowCoderTest.java         |  76 ----
 .../sql/schema/kafka/BeamKafkaCSVTableTest.java | 107 -----
 .../sql/schema/text/BeamTextCSVTableTest.java   | 176 -------
 .../transform/BeamAggregationTransformTest.java | 453 -------------------
 .../schema/transform/BeamTransformBaseTest.java |  97 ----
 92 files changed, 2575 insertions(+), 2545 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamRecordSqlType.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamRecordSqlType.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamRecordSqlType.java
new file mode 100644
index 0000000..5269867
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamRecordSqlType.java
@@ -0,0 +1,185 @@
+/*
+ * 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 java.math.BigDecimal;
+import java.sql.Types;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.GregorianCalendar;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.beam.sdk.coders.BigDecimalCoder;
+import org.apache.beam.sdk.coders.BigEndianIntegerCoder;
+import org.apache.beam.sdk.coders.BigEndianLongCoder;
+import org.apache.beam.sdk.coders.ByteCoder;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.extensions.sql.BeamSqlRecordHelper.BooleanCoder;
+import org.apache.beam.sdk.extensions.sql.BeamSqlRecordHelper.DateCoder;
+import org.apache.beam.sdk.extensions.sql.BeamSqlRecordHelper.DoubleCoder;
+import org.apache.beam.sdk.extensions.sql.BeamSqlRecordHelper.FloatCoder;
+import org.apache.beam.sdk.extensions.sql.BeamSqlRecordHelper.ShortCoder;
+import org.apache.beam.sdk.extensions.sql.BeamSqlRecordHelper.TimeCoder;
+import org.apache.beam.sdk.values.BeamRecord;
+import org.apache.beam.sdk.values.BeamRecordType;
+
+/**
+ * Type provider for {@link BeamRecord} with SQL types.
+ *
+ * <p>Limited SQL types are supported now, visit
+ * <a href="https://beam.apache.org/blog/2017/07/21/sql-dsl.html#data-type">data types</a>
+ * for more details.
+ *
+ */
+public class BeamRecordSqlType extends BeamRecordType {
+  private static final Map<Integer, Class> SQL_TYPE_TO_JAVA_CLASS = new HashMap<>();
+  static {
+    SQL_TYPE_TO_JAVA_CLASS.put(Types.TINYINT, Byte.class);
+    SQL_TYPE_TO_JAVA_CLASS.put(Types.SMALLINT, Short.class);
+    SQL_TYPE_TO_JAVA_CLASS.put(Types.INTEGER, Integer.class);
+    SQL_TYPE_TO_JAVA_CLASS.put(Types.BIGINT, Long.class);
+    SQL_TYPE_TO_JAVA_CLASS.put(Types.FLOAT, Float.class);
+    SQL_TYPE_TO_JAVA_CLASS.put(Types.DOUBLE, Double.class);
+    SQL_TYPE_TO_JAVA_CLASS.put(Types.DECIMAL, BigDecimal.class);
+
+    SQL_TYPE_TO_JAVA_CLASS.put(Types.BOOLEAN, Boolean.class);
+
+    SQL_TYPE_TO_JAVA_CLASS.put(Types.CHAR, String.class);
+    SQL_TYPE_TO_JAVA_CLASS.put(Types.VARCHAR, String.class);
+
+    SQL_TYPE_TO_JAVA_CLASS.put(Types.TIME, GregorianCalendar.class);
+
+    SQL_TYPE_TO_JAVA_CLASS.put(Types.DATE, Date.class);
+    SQL_TYPE_TO_JAVA_CLASS.put(Types.TIMESTAMP, Date.class);
+  }
+
+  public List<Integer> fieldTypes;
+
+  protected BeamRecordSqlType(List<String> fieldsName, List<Coder> fieldsCoder) {
+    super(fieldsName, fieldsCoder);
+  }
+
+  private BeamRecordSqlType(List<String> fieldsName, List<Integer> fieldTypes
+      , List<Coder> fieldsCoder) {
+    super(fieldsName, fieldsCoder);
+    this.fieldTypes = fieldTypes;
+  }
+
+  public static BeamRecordSqlType create(List<String> fieldNames,
+      List<Integer> fieldTypes) {
+    if (fieldNames.size() != fieldTypes.size()) {
+      throw new IllegalStateException("the sizes of 'dataType' and 'fieldTypes' must match.");
+    }
+    List<Coder> fieldCoders = new ArrayList<>(fieldTypes.size());
+    for (int idx = 0; idx < fieldTypes.size(); ++idx) {
+      switch (fieldTypes.get(idx)) {
+      case Types.INTEGER:
+        fieldCoders.add(BigEndianIntegerCoder.of());
+        break;
+      case Types.SMALLINT:
+        fieldCoders.add(ShortCoder.of());
+        break;
+      case Types.TINYINT:
+        fieldCoders.add(ByteCoder.of());
+        break;
+      case Types.DOUBLE:
+        fieldCoders.add(DoubleCoder.of());
+        break;
+      case Types.FLOAT:
+        fieldCoders.add(FloatCoder.of());
+        break;
+      case Types.DECIMAL:
+        fieldCoders.add(BigDecimalCoder.of());
+        break;
+      case Types.BIGINT:
+        fieldCoders.add(BigEndianLongCoder.of());
+        break;
+      case Types.VARCHAR:
+      case Types.CHAR:
+        fieldCoders.add(StringUtf8Coder.of());
+        break;
+      case Types.TIME:
+        fieldCoders.add(TimeCoder.of());
+        break;
+      case Types.DATE:
+      case Types.TIMESTAMP:
+        fieldCoders.add(DateCoder.of());
+        break;
+      case Types.BOOLEAN:
+        fieldCoders.add(BooleanCoder.of());
+        break;
+
+      default:
+        throw new UnsupportedOperationException(
+            "Data type: " + fieldTypes.get(idx) + " not supported yet!");
+      }
+    }
+    return new BeamRecordSqlType(fieldNames, fieldTypes, fieldCoders);
+  }
+
+  @Override
+  public void validateValueType(int index, Object fieldValue) throws IllegalArgumentException {
+    if (null == fieldValue) {// no need to do type check for NULL value
+      return;
+    }
+
+    int fieldType = fieldTypes.get(index);
+    Class javaClazz = SQL_TYPE_TO_JAVA_CLASS.get(fieldType);
+    if (javaClazz == null) {
+      throw new IllegalArgumentException("Data type: " + fieldType + " not supported yet!");
+    }
+
+    if (!fieldValue.getClass().equals(javaClazz)) {
+      throw new IllegalArgumentException(
+          String.format("[%s](%s) doesn't match type [%s]",
+              fieldValue, fieldValue.getClass(), fieldType)
+      );
+    }
+  }
+
+  public List<Integer> getFieldTypes() {
+    return fieldTypes;
+  }
+
+  public Integer getFieldTypeByIndex(int index){
+    return fieldTypes.get(index);
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (obj != null && obj instanceof BeamRecordSqlType) {
+      BeamRecordSqlType ins = (BeamRecordSqlType) obj;
+      return fieldTypes.equals(ins.getFieldTypes()) && getFieldNames().equals(ins.getFieldNames());
+    } else {
+      return false;
+    }
+  }
+
+  @Override
+  public int hashCode() {
+    return 31 * getFieldNames().hashCode() + getFieldTypes().hashCode();
+  }
+
+  @Override
+  public String toString() {
+    return "BeamRecordSqlType [fieldNames=" + getFieldNames()
+        + ", fieldTypes=" + fieldTypes + "]";
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSql.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSql.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSql.java
index bf6a9c0..34355fb 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSql.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSql.java
@@ -17,13 +17,11 @@
  */
 package org.apache.beam.sdk.extensions.sql;
 
-import com.google.auto.value.AutoValue;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.coders.BeamRecordCoder;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
 import org.apache.beam.sdk.extensions.sql.impl.rel.BeamRelNode;
-import org.apache.beam.sdk.extensions.sql.schema.BeamPCollectionTable;
-import org.apache.beam.sdk.extensions.sql.schema.BeamRecordSqlType;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlUdf;
+import org.apache.beam.sdk.extensions.sql.impl.schema.BeamPCollectionTable;
 import org.apache.beam.sdk.transforms.Combine.CombineFn;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.SerializableFunction;
@@ -94,10 +92,7 @@ public class BeamSql {
    * </ul>
    */
   public static QueryTransform query(String sqlQuery) {
-    return QueryTransform.builder()
-        .setSqlEnv(new BeamSqlEnv())
-        .setSqlQuery(sqlQuery)
-        .build();
+    return new QueryTransform(sqlQuery);
   }
 
   /**
@@ -109,10 +104,7 @@ public class BeamSql {
    * <p>Make sure to query it from a static table name <em>PCOLLECTION</em>.
    */
   public static SimpleQueryTransform simpleQuery(String sqlQuery) {
-    return SimpleQueryTransform.builder()
-        .setSqlEnv(new BeamSqlEnv())
-        .setSqlQuery(sqlQuery)
-        .build();
+    return new SimpleQueryTransform(sqlQuery);
   }
 
   /**
@@ -121,28 +113,22 @@ public class BeamSql {
    * <p>The table names in the input {@code PCollectionTuple} are only valid during the current
    * query.
    */
-  @AutoValue
-  public abstract static class QueryTransform extends
+  public static class QueryTransform extends
       PTransform<PCollectionTuple, PCollection<BeamRecord>> {
-    abstract BeamSqlEnv getSqlEnv();
-    abstract String getSqlQuery();
+    private BeamSqlEnv beamSqlEnv = new BeamSqlEnv();
+    private String sqlQuery;
 
-    static Builder builder() {
-      return new AutoValue_BeamSql_QueryTransform.Builder();
-    }
-
-    @AutoValue.Builder
-    abstract static class Builder {
-      abstract Builder setSqlQuery(String sqlQuery);
-      abstract Builder setSqlEnv(BeamSqlEnv sqlEnv);
-      abstract QueryTransform build();
+    public QueryTransform(String sqlQuery) {
+      this.sqlQuery = sqlQuery;
     }
 
     /**
      * register a UDF function used in this query.
+     *
+     * <p>Refer to {@link BeamSqlUdf} for more about how to implement a UDF in BeamSql.
      */
      public QueryTransform withUdf(String functionName, Class<? extends BeamSqlUdf> clazz){
-       getSqlEnv().registerUdf(functionName, clazz);
+       beamSqlEnv.registerUdf(functionName, clazz);
        return this;
      }
      /**
@@ -150,7 +136,7 @@ public class BeamSql {
       * Note, {@link SerializableFunction} must have a constructor without arguments.
       */
       public QueryTransform withUdf(String functionName, SerializableFunction sfn){
-        getSqlEnv().registerUdf(functionName, sfn);
+        beamSqlEnv.registerUdf(functionName, sfn);
         return this;
       }
 
@@ -158,7 +144,7 @@ public class BeamSql {
       * register a {@link CombineFn} as UDAF function used in this query.
       */
      public QueryTransform withUdaf(String functionName, CombineFn combineFn){
-       getSqlEnv().registerUdaf(functionName, combineFn);
+       beamSqlEnv.registerUdaf(functionName, combineFn);
        return this;
      }
 
@@ -168,13 +154,13 @@ public class BeamSql {
 
       BeamRelNode beamRelNode = null;
       try {
-        beamRelNode = getSqlEnv().planner.convertToBeamRel(getSqlQuery());
+        beamRelNode = beamSqlEnv.getPlanner().convertToBeamRel(sqlQuery);
       } catch (ValidationException | RelConversionException | SqlParseException e) {
         throw new IllegalStateException(e);
       }
 
       try {
-        return beamRelNode.buildBeamPipeline(input, getSqlEnv());
+        return beamRelNode.buildBeamPipeline(input, beamSqlEnv);
       } catch (Exception e) {
         throw new IllegalStateException(e);
       }
@@ -186,7 +172,7 @@ public class BeamSql {
         PCollection<BeamRecord> sourceStream = (PCollection<BeamRecord>) input.get(sourceTag);
         BeamRecordCoder sourceCoder = (BeamRecordCoder) sourceStream.getCoder();
 
-        getSqlEnv().registerTable(sourceTag.getId(),
+        beamSqlEnv.registerTable(sourceTag.getId(),
             new BeamPCollectionTable(sourceStream,
                 (BeamRecordSqlType) sourceCoder.getRecordType()));
       }
@@ -197,53 +183,47 @@ public class BeamSql {
    * A {@link PTransform} representing an execution plan for a SQL query referencing
    * a single table.
    */
-  @AutoValue
-  public abstract static class SimpleQueryTransform
+  public static class SimpleQueryTransform
       extends PTransform<PCollection<BeamRecord>, PCollection<BeamRecord>> {
     private static final String PCOLLECTION_TABLE_NAME = "PCOLLECTION";
-    abstract BeamSqlEnv getSqlEnv();
-    abstract String getSqlQuery();
+    private QueryTransform delegate;
 
-    static Builder builder() {
-      return new AutoValue_BeamSql_SimpleQueryTransform.Builder();
+    public SimpleQueryTransform(String sqlQuery) {
+      this.delegate = new QueryTransform(sqlQuery);
     }
 
-    @AutoValue.Builder
-    abstract static class Builder {
-      abstract Builder setSqlQuery(String sqlQuery);
-      abstract Builder setSqlEnv(BeamSqlEnv sqlEnv);
-      abstract SimpleQueryTransform build();
+    /**
+     * register a UDF function used in this query.
+     *
+     * <p>Refer to {@link BeamSqlUdf} for more about how to implement a UDAF in BeamSql.
+     */
+    public SimpleQueryTransform withUdf(String functionName, Class<? extends BeamSqlUdf> clazz){
+      delegate.withUdf(functionName, clazz);
+      return this;
     }
 
     /**
-     * register a UDF function used in this query.
+     * register {@link SerializableFunction} as a UDF function used in this query.
+     * Note, {@link SerializableFunction} must have a constructor without arguments.
      */
-     public SimpleQueryTransform withUdf(String functionName, Class<? extends BeamSqlUdf> clazz){
-       getSqlEnv().registerUdf(functionName, clazz);
-       return this;
-     }
-     /**
-      * register {@link SerializableFunction} as a UDF function used in this query.
-      * Note, {@link SerializableFunction} must have a constructor without arguments.
-      */
-      public SimpleQueryTransform withUdf(String functionName, SerializableFunction sfn){
-        getSqlEnv().registerUdf(functionName, sfn);
-        return this;
-      }
+    public SimpleQueryTransform withUdf(String functionName, SerializableFunction sfn){
+      delegate.withUdf(functionName, sfn);
+      return this;
+    }
 
-      /**
-       * register a {@link CombineFn} as UDAF function used in this query.
-       */
-      public SimpleQueryTransform withUdaf(String functionName, CombineFn combineFn){
-        getSqlEnv().registerUdaf(functionName, combineFn);
-        return this;
-      }
+    /**
+     * register a {@link CombineFn} as UDAF function used in this query.
+     */
+    public SimpleQueryTransform withUdaf(String functionName, CombineFn combineFn){
+      delegate.withUdaf(functionName, combineFn);
+      return this;
+    }
 
     private void validateQuery() {
       SqlNode sqlNode;
       try {
-        sqlNode = getSqlEnv().planner.parseQuery(getSqlQuery());
-        getSqlEnv().planner.getPlanner().close();
+        sqlNode = delegate.beamSqlEnv.getPlanner().parseQuery(delegate.sqlQuery);
+        delegate.beamSqlEnv.getPlanner().getPlanner().close();
       } catch (SqlParseException e) {
         throw new IllegalStateException(e);
       }
@@ -264,10 +244,7 @@ public class BeamSql {
     public PCollection<BeamRecord> expand(PCollection<BeamRecord> input) {
       validateQuery();
       return PCollectionTuple.of(new TupleTag<BeamRecord>(PCOLLECTION_TABLE_NAME), input)
-          .apply(QueryTransform.builder()
-              .setSqlEnv(getSqlEnv())
-              .setSqlQuery(getSqlQuery())
-              .build());
+          .apply(delegate);
     }
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlCli.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlCli.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlCli.java
deleted file mode 100644
index a43808e..0000000
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlCli.java
+++ /dev/null
@@ -1,65 +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;
-
-import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.annotations.Experimental;
-import org.apache.beam.sdk.extensions.sql.impl.rel.BeamRelNode;
-import org.apache.beam.sdk.options.PipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.apache.beam.sdk.values.BeamRecord;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.calcite.plan.RelOptUtil;
-
-/**
- * {@link BeamSqlCli} provides methods to execute Beam SQL with an interactive client.
- */
-@Experimental
-public class BeamSqlCli {
-  /**
-   * Returns a human readable representation of the query execution plan.
-   */
-  public static String explainQuery(String sqlString, BeamSqlEnv sqlEnv) throws Exception {
-    BeamRelNode exeTree = sqlEnv.planner.convertToBeamRel(sqlString);
-    String beamPlan = RelOptUtil.toString(exeTree);
-    return beamPlan;
-  }
-
-  /**
-   * compile SQL, and return a {@link Pipeline}.
-   */
-  public static PCollection<BeamRecord> compilePipeline(String sqlStatement, BeamSqlEnv sqlEnv)
-      throws Exception{
-    PipelineOptions options = PipelineOptionsFactory.fromArgs(new String[] {}).withValidation()
-        .as(PipelineOptions.class); // FlinkPipelineOptions.class
-    options.setJobName("BeamPlanCreator");
-    Pipeline pipeline = Pipeline.create(options);
-
-    return compilePipeline(sqlStatement, pipeline, sqlEnv);
-  }
-
-  /**
-   * compile SQL, and return a {@link Pipeline}.
-   */
-  public static PCollection<BeamRecord> compilePipeline(String sqlStatement, Pipeline basePipeline,
-      BeamSqlEnv sqlEnv) throws Exception{
-    PCollection<BeamRecord> resultStream =
-        sqlEnv.planner.compileBeamPipeline(sqlStatement, basePipeline, sqlEnv);
-    return resultStream;
-  }
-}

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlEnv.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlEnv.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlEnv.java
deleted file mode 100644
index 79f2b32..0000000
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlEnv.java
+++ /dev/null
@@ -1,127 +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;
-
-import java.io.Serializable;
-import org.apache.beam.sdk.extensions.sql.impl.interpreter.operator.UdafImpl;
-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.extensions.sql.schema.BaseBeamTable;
-import org.apache.beam.sdk.extensions.sql.schema.BeamRecordSqlType;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlUdf;
-import org.apache.beam.sdk.transforms.Combine.CombineFn;
-import org.apache.beam.sdk.transforms.SerializableFunction;
-import org.apache.calcite.DataContext;
-import org.apache.calcite.linq4j.Enumerable;
-import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.rel.type.RelDataTypeFactory;
-import org.apache.calcite.schema.ScannableTable;
-import org.apache.calcite.schema.Schema;
-import org.apache.calcite.schema.SchemaPlus;
-import org.apache.calcite.schema.Statistic;
-import org.apache.calcite.schema.Statistics;
-import org.apache.calcite.schema.impl.ScalarFunctionImpl;
-import org.apache.calcite.tools.Frameworks;
-
-/**
- * {@link BeamSqlEnv} prepares the execution context for {@link BeamSql} and {@link BeamSqlCli}.
- *
- * <p>It contains a {@link SchemaPlus} which holds the metadata of tables/UDF functions, and
- * a {@link BeamQueryPlanner} which parse/validate/optimize/translate input SQL queries.
- */
-public class BeamSqlEnv implements Serializable{
-  transient SchemaPlus schema;
-  transient BeamQueryPlanner planner;
-
-  public BeamSqlEnv() {
-    schema = Frameworks.createRootSchema(true);
-    planner = new BeamQueryPlanner(schema);
-  }
-
-  /**
-   * Register a UDF function which can be used in SQL expression.
-   */
-  public void registerUdf(String functionName, Class<? extends BeamSqlUdf> clazz) {
-    schema.add(functionName, ScalarFunctionImpl.create(clazz, BeamSqlUdf.UDF_METHOD));
-  }
-
-  /**
-   * register {@link SerializableFunction} as a UDF function which can be used in SQL expression.
-   * Note, {@link SerializableFunction} must have a constructor without arguments.
-   */
-  public void registerUdf(String functionName, SerializableFunction sfn) {
-    schema.add(functionName, ScalarFunctionImpl.create(sfn.getClass(), "apply"));
-  }
-
-  /**
-   * Register a {@link CombineFn} as UDAF function which can be used in GROUP-BY expression.
-   */
-  public void registerUdaf(String functionName, CombineFn combineFn) {
-    schema.add(functionName, new UdafImpl(combineFn));
-  }
-
-  /**
-   * Registers a {@link BaseBeamTable} which can be used for all subsequent queries.
-   *
-   */
-  public void registerTable(String tableName, BaseBeamTable table) {
-    schema.add(tableName, new BeamCalciteTable(table.getRowType()));
-    planner.getSourceTables().put(tableName, table);
-  }
-
-  /**
-   * Find {@link BaseBeamTable} by table name.
-   */
-  public BaseBeamTable findTable(String tableName){
-    return planner.getSourceTables().get(tableName);
-  }
-
-  private static class BeamCalciteTable implements ScannableTable, Serializable {
-    private BeamRecordSqlType beamSqlRowType;
-    public BeamCalciteTable(BeamRecordSqlType beamSqlRowType) {
-      this.beamSqlRowType = beamSqlRowType;
-    }
-    @Override
-    public RelDataType getRowType(RelDataTypeFactory typeFactory) {
-      return CalciteUtils.toCalciteRowType(this.beamSqlRowType)
-          .apply(BeamQueryPlanner.TYPE_FACTORY);
-    }
-
-    @Override
-    public Enumerable<Object[]> scan(DataContext root) {
-      // not used as Beam SQL uses its own execution engine
-      return null;
-    }
-
-    /**
-     * Not used {@link Statistic} to optimize the plan.
-     */
-    @Override
-    public Statistic getStatistic() {
-      return Statistics.UNKNOWN;
-    }
-
-    /**
-     * all sources are treated as TABLE in Beam SQL.
-     */
-    @Override
-    public Schema.TableType getJdbcTableType() {
-      return Schema.TableType.TABLE;
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlRecordHelper.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlRecordHelper.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlRecordHelper.java
new file mode 100644
index 0000000..870165d
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlRecordHelper.java
@@ -0,0 +1,217 @@
+/*
+ * 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 java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.math.BigDecimal;
+import java.util.Date;
+import java.util.GregorianCalendar;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.coders.BigDecimalCoder;
+import org.apache.beam.sdk.coders.BigEndianLongCoder;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.values.BeamRecord;
+
+/**
+ * A {@link Coder} encodes {@link BeamRecord}.
+ */
+@Experimental
+public class BeamSqlRecordHelper {
+
+  public static BeamRecordSqlType getSqlRecordType(BeamRecord record) {
+    return (BeamRecordSqlType) record.getDataType();
+  }
+
+  /**
+   * {@link Coder} for Java type {@link Short}.
+   */
+  public static class ShortCoder extends CustomCoder<Short> {
+    private static final ShortCoder INSTANCE = new ShortCoder();
+
+    public static ShortCoder of() {
+      return INSTANCE;
+    }
+
+    private ShortCoder() {
+    }
+
+    @Override
+    public void encode(Short value, OutputStream outStream) throws CoderException, IOException {
+      new DataOutputStream(outStream).writeShort(value);
+    }
+
+    @Override
+    public Short decode(InputStream inStream) throws CoderException, IOException {
+      return new DataInputStream(inStream).readShort();
+    }
+
+    @Override
+    public void verifyDeterministic() throws NonDeterministicException {
+    }
+  }
+  /**
+   * {@link Coder} for Java type {@link Float}, it's stored as {@link BigDecimal}.
+   */
+  public static class FloatCoder extends CustomCoder<Float> {
+    private static final FloatCoder INSTANCE = new FloatCoder();
+    private static final BigDecimalCoder CODER = BigDecimalCoder.of();
+
+    public static FloatCoder of() {
+      return INSTANCE;
+    }
+
+    private FloatCoder() {
+    }
+
+    @Override
+    public void encode(Float value, OutputStream outStream) throws CoderException, IOException {
+      CODER.encode(new BigDecimal(value), outStream);
+    }
+
+    @Override
+    public Float decode(InputStream inStream) throws CoderException, IOException {
+      return CODER.decode(inStream).floatValue();
+    }
+
+    @Override
+    public void verifyDeterministic() throws NonDeterministicException {
+    }
+  }
+  /**
+   * {@link Coder} for Java type {@link Double}, it's stored as {@link BigDecimal}.
+   */
+  public static class DoubleCoder extends CustomCoder<Double> {
+    private static final DoubleCoder INSTANCE = new DoubleCoder();
+    private static final BigDecimalCoder CODER = BigDecimalCoder.of();
+
+    public static DoubleCoder of() {
+      return INSTANCE;
+    }
+
+    private DoubleCoder() {
+    }
+
+    @Override
+    public void encode(Double value, OutputStream outStream) throws CoderException, IOException {
+      CODER.encode(new BigDecimal(value), outStream);
+    }
+
+    @Override
+    public Double decode(InputStream inStream) throws CoderException, IOException {
+      return CODER.decode(inStream).doubleValue();
+    }
+
+    @Override
+    public void verifyDeterministic() throws NonDeterministicException {
+    }
+  }
+
+  /**
+   * {@link Coder} for Java type {@link GregorianCalendar}, it's stored as {@link Long}.
+   */
+  public static class TimeCoder extends CustomCoder<GregorianCalendar> {
+    private static final BigEndianLongCoder longCoder = BigEndianLongCoder.of();
+    private static final TimeCoder INSTANCE = new TimeCoder();
+
+    public static TimeCoder of() {
+      return INSTANCE;
+    }
+
+    private TimeCoder() {
+    }
+
+    @Override
+    public void encode(GregorianCalendar value, OutputStream outStream)
+        throws CoderException, IOException {
+      longCoder.encode(value.getTime().getTime(), outStream);
+    }
+
+    @Override
+    public GregorianCalendar decode(InputStream inStream) throws CoderException, IOException {
+      GregorianCalendar calendar = new GregorianCalendar();
+      calendar.setTime(new Date(longCoder.decode(inStream)));
+      return calendar;
+    }
+
+    @Override
+    public void verifyDeterministic() throws NonDeterministicException {
+    }
+  }
+  /**
+   * {@link Coder} for Java type {@link Date}, it's stored as {@link Long}.
+   */
+  public static class DateCoder extends CustomCoder<Date> {
+    private static final BigEndianLongCoder longCoder = BigEndianLongCoder.of();
+    private static final DateCoder INSTANCE = new DateCoder();
+
+    public static DateCoder of() {
+      return INSTANCE;
+    }
+
+    private DateCoder() {
+    }
+
+    @Override
+    public void encode(Date value, OutputStream outStream) throws CoderException, IOException {
+      longCoder.encode(value.getTime(), outStream);
+    }
+
+    @Override
+    public Date decode(InputStream inStream) throws CoderException, IOException {
+      return new Date(longCoder.decode(inStream));
+    }
+
+    @Override
+    public void verifyDeterministic() throws NonDeterministicException {
+    }
+  }
+
+  /**
+   * {@link Coder} for Java type {@link Boolean}.
+   */
+  public static class BooleanCoder extends CustomCoder<Boolean> {
+    private static final BooleanCoder INSTANCE = new BooleanCoder();
+
+    public static BooleanCoder of() {
+      return INSTANCE;
+    }
+
+    private BooleanCoder() {
+    }
+
+    @Override
+    public void encode(Boolean value, OutputStream outStream) throws CoderException, IOException {
+      new DataOutputStream(outStream).writeBoolean(value);
+    }
+
+    @Override
+    public Boolean decode(InputStream inStream) throws CoderException, IOException {
+      return new DataInputStream(inStream).readBoolean();
+    }
+
+    @Override
+    public void verifyDeterministic() throws NonDeterministicException {
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlUdf.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlUdf.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlUdf.java
new file mode 100644
index 0000000..d4828e7
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/BeamSqlUdf.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;
+
+import java.io.Serializable;
+
+/**
+ * Interface to create a UDF in Beam SQL.
+ *
+ * <p>A static method {@code eval} is required. Here is an example:
+ *
+ * <blockquote><pre>
+ * public static class MyLeftFunction {
+ *   public String eval(
+ *       &#64;Parameter(name = "s") String s,
+ *       &#64;Parameter(name = "n", optional = true) Integer n) {
+ *     return s.substring(0, n == null ? 1 : n);
+ *   }
+ * }</pre></blockquote>
+ *
+ * <p>The first parameter is named "s" and is mandatory,
+ * and the second parameter is named "n" and is optional.
+ */
+public interface BeamSqlUdf extends Serializable {
+  String UDF_METHOD = "eval";
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
index 91251cf..0c5dae1 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
@@ -21,8 +21,8 @@ import java.sql.Types;
 import java.util.Arrays;
 import java.util.List;
 import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
 import org.apache.beam.sdk.extensions.sql.BeamSql;
-import org.apache.beam.sdk.extensions.sql.schema.BeamRecordSqlType;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.transforms.Create;

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/BeamSqlCli.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/BeamSqlCli.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/BeamSqlCli.java
new file mode 100644
index 0000000..5c7d920
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/BeamSqlCli.java
@@ -0,0 +1,65 @@
+/*
+ * 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;
+
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.extensions.sql.impl.rel.BeamRelNode;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.values.BeamRecord;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.calcite.plan.RelOptUtil;
+
+/**
+ * {@link BeamSqlCli} provides methods to execute Beam SQL with an interactive client.
+ */
+@Experimental
+public class BeamSqlCli {
+  /**
+   * Returns a human readable representation of the query execution plan.
+   */
+  public static String explainQuery(String sqlString, BeamSqlEnv sqlEnv) throws Exception {
+    BeamRelNode exeTree = sqlEnv.getPlanner().convertToBeamRel(sqlString);
+    String beamPlan = RelOptUtil.toString(exeTree);
+    return beamPlan;
+  }
+
+  /**
+   * compile SQL, and return a {@link Pipeline}.
+   */
+  public static PCollection<BeamRecord> compilePipeline(String sqlStatement, BeamSqlEnv sqlEnv)
+      throws Exception{
+    PipelineOptions options = PipelineOptionsFactory.fromArgs(new String[] {}).withValidation()
+        .as(PipelineOptions.class);
+    options.setJobName("BeamPlanCreator");
+    Pipeline pipeline = Pipeline.create(options);
+
+    return compilePipeline(sqlStatement, pipeline, sqlEnv);
+  }
+
+  /**
+   * compile SQL, and return a {@link Pipeline}.
+   */
+  public static PCollection<BeamRecord> compilePipeline(String sqlStatement, Pipeline basePipeline,
+      BeamSqlEnv sqlEnv) throws Exception{
+    PCollection<BeamRecord> resultStream = sqlEnv.getPlanner()
+        .compileBeamPipeline(sqlStatement, basePipeline, sqlEnv);
+    return resultStream;
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/BeamSqlEnv.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/BeamSqlEnv.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/BeamSqlEnv.java
new file mode 100644
index 0000000..fcc9079
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/BeamSqlEnv.java
@@ -0,0 +1,135 @@
+/*
+ * 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;
+
+import java.io.Serializable;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.extensions.sql.BeamSql;
+import org.apache.beam.sdk.extensions.sql.BeamSqlUdf;
+import org.apache.beam.sdk.extensions.sql.impl.interpreter.operator.UdafImpl;
+import org.apache.beam.sdk.extensions.sql.impl.planner.BeamQueryPlanner;
+import org.apache.beam.sdk.extensions.sql.impl.schema.BaseBeamTable;
+import org.apache.beam.sdk.extensions.sql.impl.schema.BeamSqlTable;
+import org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+import org.apache.calcite.DataContext;
+import org.apache.calcite.linq4j.Enumerable;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.schema.ScannableTable;
+import org.apache.calcite.schema.Schema;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.schema.Statistic;
+import org.apache.calcite.schema.Statistics;
+import org.apache.calcite.schema.impl.ScalarFunctionImpl;
+import org.apache.calcite.tools.Frameworks;
+
+/**
+ * {@link BeamSqlEnv} prepares the execution context for {@link BeamSql} and
+ * {@link BeamSqlCli}.
+ *
+ * <p>It contains a {@link SchemaPlus} which holds the metadata of tables/UDF functions,
+ * and a {@link BeamQueryPlanner} which parse/validate/optimize/translate input SQL queries.
+ */
+public class BeamSqlEnv implements Serializable{
+  transient SchemaPlus schema;
+  transient BeamQueryPlanner planner;
+
+  public BeamSqlEnv() {
+    schema = Frameworks.createRootSchema(true);
+    planner = new BeamQueryPlanner(schema);
+  }
+
+  /**
+   * Register a UDF function which can be used in SQL expression.
+   */
+  public void registerUdf(String functionName, Class<? extends BeamSqlUdf> clazz) {
+    schema.add(functionName, ScalarFunctionImpl.create(clazz, BeamSqlUdf.UDF_METHOD));
+  }
+
+  /**
+   * Register {@link SerializableFunction} as a UDF function which can be used in SQL expression.
+   * Note, {@link SerializableFunction} must have a constructor without arguments.
+   */
+  public void registerUdf(String functionName, SerializableFunction sfn) {
+    schema.add(functionName, ScalarFunctionImpl.create(sfn.getClass(), "apply"));
+  }
+
+  /**
+   * Register a UDAF function which can be used in GROUP-BY expression.
+   * See {@link org.apache.beam.sdk.transforms.Combine.CombineFn} on how to implement a UDAF.
+   */
+  public void registerUdaf(String functionName, Combine.CombineFn combineFn) {
+    schema.add(functionName, new UdafImpl(combineFn));
+  }
+
+  /**
+   * Registers a {@link BaseBeamTable} which can be used for all subsequent queries.
+   *
+   */
+  public void registerTable(String tableName, BeamSqlTable table) {
+    schema.add(tableName, new BeamCalciteTable(table.getRowType()));
+    planner.getSourceTables().put(tableName, table);
+  }
+
+  /**
+   * Find {@link BaseBeamTable} by table name.
+   */
+  public BeamSqlTable findTable(String tableName){
+    return planner.getSourceTables().get(tableName);
+  }
+
+  private static class BeamCalciteTable implements ScannableTable, Serializable {
+    private BeamRecordSqlType beamSqlRowType;
+    public BeamCalciteTable(BeamRecordSqlType beamSqlRowType) {
+      this.beamSqlRowType = beamSqlRowType;
+    }
+    @Override
+    public RelDataType getRowType(RelDataTypeFactory typeFactory) {
+      return CalciteUtils.toCalciteRowType(this.beamSqlRowType)
+          .apply(BeamQueryPlanner.TYPE_FACTORY);
+    }
+
+    @Override
+    public Enumerable<Object[]> scan(DataContext root) {
+      // not used as Beam SQL uses its own execution engine
+      return null;
+    }
+
+    /**
+     * Not used {@link Statistic} to optimize the plan.
+     */
+    @Override
+    public Statistic getStatistic() {
+      return Statistics.UNKNOWN;
+    }
+
+    /**
+     * all sources are treated as TABLE in Beam SQL.
+     */
+    @Override
+    public Schema.TableType getJdbcTableType() {
+      return Schema.TableType.TABLE;
+    }
+  }
+
+  public BeamQueryPlanner getPlanner() {
+    return planner;
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/package-info.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/package-info.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/package-info.java
new file mode 100644
index 0000000..de237d6
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/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.
+ */
+
+/**
+ * Implementation classes of BeamSql.
+ */
+package org.apache.beam.sdk.extensions.sql.impl;

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamQueryPlanner.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamQueryPlanner.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamQueryPlanner.java
index b421bc3..410c783 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamQueryPlanner.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamQueryPlanner.java
@@ -23,10 +23,11 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
 import org.apache.beam.sdk.extensions.sql.impl.rel.BeamLogicalConvention;
 import org.apache.beam.sdk.extensions.sql.impl.rel.BeamRelNode;
-import org.apache.beam.sdk.extensions.sql.schema.BaseBeamTable;
+import org.apache.beam.sdk.extensions.sql.impl.schema.BaseBeamTable;
+import org.apache.beam.sdk.extensions.sql.impl.schema.BeamSqlTable;
 import org.apache.beam.sdk.values.BeamRecord;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionTuple;
@@ -67,7 +68,7 @@ public class BeamQueryPlanner {
   private static final Logger LOG = LoggerFactory.getLogger(BeamQueryPlanner.class);
 
   protected final Planner planner;
-  private Map<String, BaseBeamTable> sourceTables = new HashMap<>();
+  private Map<String, BeamSqlTable> sourceTables = new HashMap<>();
 
   public static final JavaTypeFactory TYPE_FACTORY = new JavaTypeFactoryImpl(
       RelDataTypeSystem.DEFAULT);
@@ -156,7 +157,7 @@ public class BeamQueryPlanner {
     return planner.validate(sqlNode);
   }
 
-  public Map<String, BaseBeamTable> getSourceTables() {
+  public Map<String, BeamSqlTable> getSourceTables() {
     return sourceTables;
   }
 

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamAggregationRel.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamAggregationRel.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamAggregationRel.java
index 4b557f9..e49e79c 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamAggregationRel.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamAggregationRel.java
@@ -21,10 +21,10 @@ import java.util.ArrayList;
 import java.util.List;
 import org.apache.beam.sdk.coders.BeamRecordCoder;
 import org.apache.beam.sdk.coders.KvCoder;
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
 import org.apache.beam.sdk.extensions.sql.impl.transform.BeamAggregationTransforms;
 import org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils;
-import org.apache.beam.sdk.extensions.sql.schema.BeamRecordSqlType;
 import org.apache.beam.sdk.transforms.Combine;
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.transforms.WithKeys;

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamFilterRel.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamFilterRel.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamFilterRel.java
index 8fe5be4..9d36a47 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamFilterRel.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamFilterRel.java
@@ -17,7 +17,7 @@
  */
 package org.apache.beam.sdk.extensions.sql.impl.rel;
 
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
 import org.apache.beam.sdk.extensions.sql.impl.interpreter.BeamSqlExpressionExecutor;
 import org.apache.beam.sdk.extensions.sql.impl.interpreter.BeamSqlFnExecutor;
 import org.apache.beam.sdk.extensions.sql.impl.transform.BeamSqlFilterFn;

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIOSinkRel.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIOSinkRel.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIOSinkRel.java
index 1e3eb4c..7bb08c2 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIOSinkRel.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIOSinkRel.java
@@ -19,8 +19,8 @@ package org.apache.beam.sdk.extensions.sql.impl.rel;
 
 import com.google.common.base.Joiner;
 import java.util.List;
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
-import org.apache.beam.sdk.extensions.sql.schema.BaseBeamTable;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.impl.schema.BeamSqlTable;
 import org.apache.beam.sdk.values.BeamRecord;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionTuple;
@@ -65,7 +65,7 @@ public class BeamIOSinkRel extends TableModify implements BeamRelNode {
 
     String sourceName = Joiner.on('.').join(getTable().getQualifiedName());
 
-    BaseBeamTable targetTable = sqlEnv.findTable(sourceName);
+    BeamSqlTable targetTable = sqlEnv.findTable(sourceName);
 
     upstream.apply(stageName, targetTable.buildIOWriter());
 

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIOSourceRel.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIOSourceRel.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIOSourceRel.java
index 254f990..1e4f506 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIOSourceRel.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIOSourceRel.java
@@ -18,9 +18,9 @@
 package org.apache.beam.sdk.extensions.sql.impl.rel;
 
 import com.google.common.base.Joiner;
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.impl.schema.BeamSqlTable;
 import org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils;
-import org.apache.beam.sdk.extensions.sql.schema.BaseBeamTable;
 import org.apache.beam.sdk.values.BeamRecord;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionTuple;
@@ -53,7 +53,7 @@ public class BeamIOSourceRel extends TableScan implements BeamRelNode {
       return sourceStream;
     } else {
       //If not, the source PColection is provided with BaseBeamTable.buildIOReader().
-      BaseBeamTable sourceTable = sqlEnv.findTable(sourceName);
+      BeamSqlTable sourceTable = sqlEnv.findTable(sourceName);
       return sourceTable.buildIOReader(inputPCollections.getPipeline())
           .setCoder(CalciteUtils.toBeamRowType(getRowType()).getRecordCoder());
     }

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIntersectRel.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIntersectRel.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIntersectRel.java
index 5919329..1ffb636 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIntersectRel.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamIntersectRel.java
@@ -19,7 +19,7 @@
 package org.apache.beam.sdk.extensions.sql.impl.rel;
 
 import java.util.List;
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
 import org.apache.beam.sdk.values.BeamRecord;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionTuple;

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamJoinRel.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamJoinRel.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamJoinRel.java
index 5ac9575..cc26aa6 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamJoinRel.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamJoinRel.java
@@ -25,10 +25,10 @@ import java.util.Map;
 import java.util.Set;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.KvCoder;
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
 import org.apache.beam.sdk.extensions.sql.impl.transform.BeamJoinTransforms;
 import org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils;
-import org.apache.beam.sdk.extensions.sql.schema.BeamRecordSqlType;
 import org.apache.beam.sdk.transforms.MapElements;
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.transforms.View;

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamMinusRel.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamMinusRel.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamMinusRel.java
index b55252a..6f5dff2 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamMinusRel.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamMinusRel.java
@@ -19,7 +19,7 @@
 package org.apache.beam.sdk.extensions.sql.impl.rel;
 
 import java.util.List;
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
 import org.apache.beam.sdk.values.BeamRecord;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionTuple;

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamProjectRel.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamProjectRel.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamProjectRel.java
index b1ff629..501feb3 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamProjectRel.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamProjectRel.java
@@ -18,7 +18,7 @@
 package org.apache.beam.sdk.extensions.sql.impl.rel;
 
 import java.util.List;
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
 import org.apache.beam.sdk.extensions.sql.impl.interpreter.BeamSqlExpressionExecutor;
 import org.apache.beam.sdk.extensions.sql.impl.interpreter.BeamSqlFnExecutor;
 import org.apache.beam.sdk.extensions.sql.impl.transform.BeamSqlProjectFn;

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamRelNode.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamRelNode.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamRelNode.java
index b8b4293..9e8d46d 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamRelNode.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamRelNode.java
@@ -17,7 +17,7 @@
  */
 package org.apache.beam.sdk.extensions.sql.impl.rel;
 
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
 import org.apache.beam.sdk.values.BeamRecord;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionTuple;
@@ -33,6 +33,7 @@ public interface BeamRelNode extends RelNode {
    * {@code BeamQueryPlanner} visits it with a DFS(Depth-First-Search)
    * algorithm.
    */
-  PCollection<BeamRecord> buildBeamPipeline(PCollectionTuple inputPCollections, BeamSqlEnv sqlEnv)
+  PCollection<BeamRecord> buildBeamPipeline(
+      PCollectionTuple inputPCollections, BeamSqlEnv sqlEnv)
       throws Exception;
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSetOperatorRelBase.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSetOperatorRelBase.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSetOperatorRelBase.java
index f9cbf4f..a1f3e2b 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSetOperatorRelBase.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSetOperatorRelBase.java
@@ -20,7 +20,7 @@ package org.apache.beam.sdk.extensions.sql.impl.rel;
 
 import java.io.Serializable;
 import java.util.List;
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
 import org.apache.beam.sdk.extensions.sql.impl.transform.BeamSetOperatorsTransforms;
 import org.apache.beam.sdk.transforms.MapElements;
 import org.apache.beam.sdk.transforms.ParDo;

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSortRel.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSortRel.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSortRel.java
index 80f3c97..d658638 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSortRel.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSortRel.java
@@ -25,9 +25,9 @@ import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.List;
 import org.apache.beam.sdk.coders.ListCoder;
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.BeamSqlRecordHelper;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
 import org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordHelper;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.Flatten;
 import org.apache.beam.sdk.transforms.ParDo;

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamUnionRel.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamUnionRel.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamUnionRel.java
index 63ebdf3..85d676e 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamUnionRel.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamUnionRel.java
@@ -19,7 +19,7 @@
 package org.apache.beam.sdk.extensions.sql.impl.rel;
 
 import java.util.List;
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
 import org.apache.beam.sdk.transforms.windowing.WindowFn;
 import org.apache.beam.sdk.values.BeamRecord;
 import org.apache.beam.sdk.values.PCollection;

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamValuesRel.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamValuesRel.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamValuesRel.java
index c4caff3..d684294 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamValuesRel.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamValuesRel.java
@@ -21,10 +21,10 @@ package org.apache.beam.sdk.extensions.sql.impl.rel;
 import com.google.common.collect.ImmutableList;
 import java.util.ArrayList;
 import java.util.List;
-import org.apache.beam.sdk.extensions.sql.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.impl.schema.BeamTableUtils;
 import org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils;
-import org.apache.beam.sdk.extensions.sql.schema.BeamRecordSqlType;
-import org.apache.beam.sdk.extensions.sql.schema.BeamTableUtils;
 import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.values.BeamRecord;
 import org.apache.beam.sdk.values.PCollection;

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BaseBeamTable.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BaseBeamTable.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BaseBeamTable.java
new file mode 100644
index 0000000..73e0863
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BaseBeamTable.java
@@ -0,0 +1,35 @@
+/*
+ * 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;
+
+import java.io.Serializable;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+
+/**
+ * Each IO in Beam has one table schema, by extending {@link BaseBeamTable}.
+ */
+public abstract class BaseBeamTable implements BeamSqlTable, Serializable {
+  protected BeamRecordSqlType beamSqlRowType;
+  public BaseBeamTable(BeamRecordSqlType beamSqlRowType) {
+    this.beamSqlRowType = beamSqlRowType;
+  }
+
+  @Override public BeamRecordSqlType getRowType() {
+    return beamSqlRowType;
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamIOType.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamIOType.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamIOType.java
new file mode 100644
index 0000000..5ced467
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamIOType.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.beam.sdk.extensions.sql.impl.schema;
+
+import java.io.Serializable;
+
+/**
+ * Type as a source IO, determined whether it's a STREAMING process, or batch
+ * process.
+ */
+public enum BeamIOType implements Serializable {
+  BOUNDED, UNBOUNDED;
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamPCollectionTable.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamPCollectionTable.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamPCollectionTable.java
new file mode 100644
index 0000000..31e60e0
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamPCollectionTable.java
@@ -0,0 +1,63 @@
+/*
+ * 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;
+
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.BeamRecord;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollection.IsBounded;
+import org.apache.beam.sdk.values.PDone;
+
+/**
+ * {@code BeamPCollectionTable} converts a {@code PCollection<BeamSqlRow>} as a virtual table,
+ * then a downstream query can query directly.
+ */
+public class BeamPCollectionTable extends BaseBeamTable {
+  private BeamIOType ioType;
+  private transient PCollection<BeamRecord> upstream;
+
+  protected BeamPCollectionTable(BeamRecordSqlType beamSqlRowType) {
+    super(beamSqlRowType);
+  }
+
+  public BeamPCollectionTable(PCollection<BeamRecord> upstream,
+      BeamRecordSqlType beamSqlRowType){
+    this(beamSqlRowType);
+    ioType = upstream.isBounded().equals(IsBounded.BOUNDED)
+        ? BeamIOType.BOUNDED : BeamIOType.UNBOUNDED;
+    this.upstream = upstream;
+  }
+
+  @Override
+  public BeamIOType getSourceType() {
+    return ioType;
+  }
+
+  @Override
+  public PCollection<BeamRecord> buildIOReader(Pipeline pipeline) {
+    return upstream;
+  }
+
+  @Override
+  public PTransform<? super PCollection<BeamRecord>, PDone> buildIOWriter() {
+    throw new IllegalArgumentException("cannot use [BeamPCollectionTable] as target");
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamSqlTable.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamSqlTable.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamSqlTable.java
new file mode 100644
index 0000000..46fba59
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamSqlTable.java
@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.BeamRecord;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
+
+/**
+ * This interface defines a Beam Sql Table.
+ */
+public interface BeamSqlTable {
+  /**
+   * In Beam SQL, there's no difference between a batch query and a streaming
+   * query. {@link BeamIOType} is used to validate the sources.
+   */
+  BeamIOType getSourceType();
+
+  /**
+   * create a {@code PCollection<BeamSqlRow>} from source.
+   *
+   */
+  PCollection<BeamRecord> buildIOReader(Pipeline pipeline);
+
+  /**
+   * create a {@code IO.write()} instance to write to target.
+   *
+   */
+   PTransform<? super PCollection<BeamRecord>, PDone> buildIOWriter();
+
+  /**
+   * Get the schema info of the table.
+   */
+   BeamRecordSqlType getRowType();
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/b17cf303/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamTableUtils.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamTableUtils.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamTableUtils.java
new file mode 100644
index 0000000..6f7f09b
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamTableUtils.java
@@ -0,0 +1,118 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.io.StringWriter;
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.beam.sdk.extensions.sql.BeamRecordSqlType;
+import org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils;
+import org.apache.beam.sdk.values.BeamRecord;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.util.NlsString;
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVParser;
+import org.apache.commons.csv.CSVPrinter;
+import org.apache.commons.csv.CSVRecord;
+
+/**
+ * Utility methods for working with {@code BeamTable}.
+ */
+public final class BeamTableUtils {
+  public static BeamRecord csvLine2BeamSqlRow(
+      CSVFormat csvFormat,
+      String line,
+      BeamRecordSqlType beamRecordSqlType) {
+    List<Object> fieldsValue = new ArrayList<>(beamRecordSqlType.getFieldCount());
+    try (StringReader reader = new StringReader(line)) {
+      CSVParser parser = csvFormat.parse(reader);
+      CSVRecord rawRecord = parser.getRecords().get(0);
+
+      if (rawRecord.size() != beamRecordSqlType.getFieldCount()) {
+        throw new IllegalArgumentException(String.format(
+            "Expect %d fields, but actually %d",
+            beamRecordSqlType.getFieldCount(), rawRecord.size()
+        ));
+      } else {
+        for (int idx = 0; idx < beamRecordSqlType.getFieldCount(); idx++) {
+          String raw = rawRecord.get(idx);
+          fieldsValue.add(autoCastField(beamRecordSqlType.getFieldTypeByIndex(idx), raw));
+        }
+      }
+    } catch (IOException e) {
+      throw new IllegalArgumentException("decodeRecord failed!", e);
+    }
+    return new BeamRecord(beamRecordSqlType, fieldsValue);
+  }
+
+  public static String beamSqlRow2CsvLine(BeamRecord row, CSVFormat csvFormat) {
+    StringWriter writer = new StringWriter();
+    try (CSVPrinter printer = csvFormat.print(writer)) {
+      for (int i = 0; i < row.getFieldCount(); i++) {
+        printer.print(row.getFieldValue(i).toString());
+      }
+      printer.println();
+    } catch (IOException e) {
+      throw new IllegalArgumentException("encodeRecord failed!", e);
+    }
+    return writer.toString();
+  }
+
+  public static Object autoCastField(int fieldType, Object rawObj) {
+    if (rawObj == null) {
+      return null;
+    }
+
+    SqlTypeName columnType = CalciteUtils.toCalciteType(fieldType);
+    // auto-casting for numberics
+    if ((rawObj instanceof String && SqlTypeName.NUMERIC_TYPES.contains(columnType))
+        || (rawObj instanceof BigDecimal && columnType != SqlTypeName.DECIMAL)) {
+      String raw = rawObj.toString();
+      switch (columnType) {
+        case TINYINT:
+          return Byte.valueOf(raw);
+        case SMALLINT:
+          return Short.valueOf(raw);
+        case INTEGER:
+          return Integer.valueOf(raw);
+        case BIGINT:
+          return Long.valueOf(raw);
+        case FLOAT:
+          return Float.valueOf(raw);
+        case DOUBLE:
+          return Double.valueOf(raw);
+        default:
+          throw new UnsupportedOperationException(
+              String.format("Column type %s is not supported yet!", columnType));
+      }
+    } else if (SqlTypeName.CHAR_TYPES.contains(columnType)) {
+      // convert NlsString to String
+      if (rawObj instanceof NlsString) {
+        return ((NlsString) rawObj).getValue();
+      } else {
+        return rawObj;
+      }
+    } else {
+      return rawObj;
+    }
+  }
+}


Mime
View raw message