[4/7] beam git commit: refactor BeamRecord, BeamRecordType, BeamSqlRecordType, BeamRecordCoder

[ro...@apache.org]

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRecordHelper.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRecordHelper.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRecordHelper.java
new file mode 100644
index 0000000..b910c84
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/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.schema;
+
+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 BeamSqlRecordType getSqlRecordType(BeamRecord record) {
+    return (BeamSqlRecordType) 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/89109b8c/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRecordType.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRecordType.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRecordType.java
new file mode 100644
index 0000000..b295049
--- /dev/null
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRecordType.java
@@ -0,0 +1,168 @@
+/*
+ * 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.schema;
+
+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.schema.BeamSqlRecordHelper.BooleanCoder;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordHelper.DateCoder;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordHelper.DoubleCoder;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordHelper.FloatCoder;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordHelper.ShortCoder;
+import org.apache.beam.sdk.extensions.sql.schema.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 BeamSqlRecordType 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> fieldsType;
+
+  protected BeamSqlRecordType(List<String> fieldsName, List<Coder> fieldsCoder) {
+    super(fieldsName, fieldsCoder);
+  }
+
+  private BeamSqlRecordType(List<String> fieldsName, List<Integer> fieldsType
+      , List<Coder> fieldsCoder) {
+    super(fieldsName, fieldsCoder);
+    this.fieldsType = fieldsType;
+  }
+
+  public static BeamSqlRecordType create(List<String> fieldNames,
+      List<Integer> fieldTypes) {
+    List<Coder> fieldCoders = new ArrayList<>();
+    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 BeamSqlRecordType(fieldNames, fieldTypes, fieldCoders);
+  }
+
+  @Override
+  public void validateValueType(int index, Object fieldValue) throws IllegalArgumentException {
+    int fieldType = fieldsType.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> getFieldsType() {
+    return fieldsType;
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (obj != null && obj instanceof BeamSqlRecordType) {
+      BeamSqlRecordType ins = (BeamSqlRecordType) obj;
+      return fieldsType.equals(ins.getFieldsType()) && getFieldsName().equals(ins.getFieldsName());
+    } else {
+      return false;
+    }
+  }
+
+  @Override
+  public int hashCode() {
+    return 31 * getFieldsName().hashCode() + getFieldsType().hashCode();
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRow.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRow.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRow.java
deleted file mode 100644
index cb5c7ea..0000000
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRow.java
+++ /dev/null
@@ -1,41 +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.schema;
-
-import java.util.List;
-import org.apache.beam.sdk.values.BeamRecord;
-import org.apache.beam.sdk.values.PCollection;
-
-/**
- * {@link BeamSqlRow} represents one row element in a {@link PCollection},
- * with type provider {@link BeamSqlRowType}.
- */
-public class BeamSqlRow extends BeamRecord {
-  public BeamSqlRow(BeamSqlRowType dataType, List<Object> dataValues) {
-    super(dataType, dataValues);
-  }
-
-  public BeamSqlRow(BeamSqlRowType dataType) {
-    super(dataType);
-  }
-
-  @Override
-  public BeamSqlRowType getDataType() {
-    return (BeamSqlRowType) super.getDataType();
-  }
-}

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRowCoder.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRowCoder.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRowCoder.java
deleted file mode 100644
index c7656af..0000000
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRowCoder.java
+++ /dev/null
@@ -1,186 +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.schema;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.sql.Types;
-import java.util.BitSet;
-import java.util.Date;
-import java.util.GregorianCalendar;
-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.BitSetCoder;
-import org.apache.beam.sdk.coders.ByteCoder;
-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.coders.DoubleCoder;
-import org.apache.beam.sdk.coders.InstantCoder;
-import org.apache.beam.sdk.coders.StringUtf8Coder;
-
-/**
- *  A {@link Coder} encodes {@link BeamSqlRow}.
- */
-public class BeamSqlRowCoder extends CustomCoder<BeamSqlRow> {
-  private BeamSqlRowType sqlRecordType;
-
-  private static final BitSetCoder nullListCoder = BitSetCoder.of();
-
-  private static final StringUtf8Coder stringCoder = StringUtf8Coder.of();
-  private static final BigEndianIntegerCoder intCoder = BigEndianIntegerCoder.of();
-  private static final BigEndianLongCoder longCoder = BigEndianLongCoder.of();
-  private static final DoubleCoder doubleCoder = DoubleCoder.of();
-  private static final InstantCoder instantCoder = InstantCoder.of();
-  private static final BigDecimalCoder bigDecimalCoder = BigDecimalCoder.of();
-  private static final ByteCoder byteCoder = ByteCoder.of();
-
-  public BeamSqlRowCoder(BeamSqlRowType sqlRecordType) {
-    this.sqlRecordType = sqlRecordType;
-  }
-
-  @Override
-  public void encode(BeamSqlRow value, OutputStream outStream)
-      throws CoderException, IOException {
-    nullListCoder.encode(value.getNullFields(), outStream);
-    for (int idx = 0; idx < value.size(); ++idx) {
-      if (value.getNullFields().get(idx)) {
-        continue;
-      }
-
-      switch (sqlRecordType.getFieldsType().get(idx)) {
-        case Types.INTEGER:
-          intCoder.encode(value.getInteger(idx), outStream);
-          break;
-        case Types.SMALLINT:
-          intCoder.encode((int) value.getShort(idx), outStream);
-          break;
-        case Types.TINYINT:
-          byteCoder.encode(value.getByte(idx), outStream);
-          break;
-        case Types.DOUBLE:
-          doubleCoder.encode(value.getDouble(idx), outStream);
-          break;
-        case Types.FLOAT:
-          doubleCoder.encode((double) value.getFloat(idx), outStream);
-          break;
-        case Types.DECIMAL:
-          bigDecimalCoder.encode(value.getBigDecimal(idx), outStream);
-          break;
-        case Types.BIGINT:
-          longCoder.encode(value.getLong(idx), outStream);
-          break;
-        case Types.VARCHAR:
-        case Types.CHAR:
-          stringCoder.encode(value.getString(idx), outStream);
-          break;
-        case Types.TIME:
-          longCoder.encode(value.getGregorianCalendar(idx).getTime().getTime(), outStream);
-          break;
-        case Types.DATE:
-        case Types.TIMESTAMP:
-          longCoder.encode(value.getDate(idx).getTime(), outStream);
-          break;
-        case Types.BOOLEAN:
-          byteCoder.encode((byte) (value.getBoolean(idx) ? 1 : 0), outStream);
-          break;
-
-        default:
-          throw new UnsupportedOperationException(
-              "Data type: " + sqlRecordType.getFieldsType().get(idx) + " not supported yet!");
-      }
-    }
-
-    instantCoder.encode(value.getWindowStart(), outStream);
-    instantCoder.encode(value.getWindowEnd(), outStream);
-  }
-
-  @Override
-  public BeamSqlRow decode(InputStream inStream) throws CoderException, IOException {
-    BitSet nullFields = nullListCoder.decode(inStream);
-
-    BeamSqlRow record = new BeamSqlRow(sqlRecordType);
-    record.setNullFields(nullFields);
-    for (int idx = 0; idx < sqlRecordType.size(); ++idx) {
-      if (nullFields.get(idx)) {
-        continue;
-      }
-
-      switch (sqlRecordType.getFieldsType().get(idx)) {
-        case Types.INTEGER:
-          record.addField(idx, intCoder.decode(inStream));
-          break;
-        case Types.SMALLINT:
-          record.addField(idx, intCoder.decode(inStream).shortValue());
-          break;
-        case Types.TINYINT:
-          record.addField(idx, byteCoder.decode(inStream));
-          break;
-        case Types.DOUBLE:
-          record.addField(idx, doubleCoder.decode(inStream));
-          break;
-        case Types.FLOAT:
-          record.addField(idx, doubleCoder.decode(inStream).floatValue());
-          break;
-        case Types.BIGINT:
-          record.addField(idx, longCoder.decode(inStream));
-          break;
-        case Types.DECIMAL:
-          record.addField(idx, bigDecimalCoder.decode(inStream));
-          break;
-        case Types.VARCHAR:
-        case Types.CHAR:
-          record.addField(idx, stringCoder.decode(inStream));
-          break;
-        case Types.TIME:
-          GregorianCalendar calendar = new GregorianCalendar();
-          calendar.setTime(new Date(longCoder.decode(inStream)));
-          record.addField(idx, calendar);
-          break;
-        case Types.DATE:
-        case Types.TIMESTAMP:
-          record.addField(idx, new Date(longCoder.decode(inStream)));
-          break;
-        case Types.BOOLEAN:
-          record.addField(idx, byteCoder.decode(inStream) == 1);
-          break;
-
-        default:
-          throw new UnsupportedOperationException("Data type: "
-              + sqlRecordType.getFieldsType().get(idx)
-              + " not supported yet!");
-      }
-    }
-
-    record.setWindowStart(instantCoder.decode(inStream));
-    record.setWindowEnd(instantCoder.decode(inStream));
-
-    return record;
-  }
-
-  public BeamSqlRowType getSqlRecordType() {
-    return sqlRecordType;
-  }
-
-  @Override
-  public void verifyDeterministic()
-      throws org.apache.beam.sdk.coders.Coder.NonDeterministicException {
-  }
-}

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRowType.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRowType.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRowType.java
deleted file mode 100644
index 7584dad..0000000
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlRowType.java
+++ /dev/null
@@ -1,109 +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.schema;
-
-import java.math.BigDecimal;
-import java.sql.Types;
-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.values.BeamRecordTypeProvider;
-
-/**
- * Type provider for {@link BeamSqlRow} 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 BeamSqlRowType extends BeamRecordTypeProvider {
-  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> fieldsType;
-
-  protected BeamSqlRowType(List<String> fieldsName) {
-    super(fieldsName);
-  }
-
-  public BeamSqlRowType(List<String> fieldsName, List<Integer> fieldsType) {
-    super(fieldsName);
-    this.fieldsType = fieldsType;
-  }
-
-  public static BeamSqlRowType create(List<String> fieldNames,
-      List<Integer> fieldTypes) {
-    return new BeamSqlRowType(fieldNames, fieldTypes);
-  }
-
-  @Override
-  public void validateValueType(int index, Object fieldValue) throws IllegalArgumentException {
-    int fieldType = fieldsType.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> getFieldsType() {
-    return fieldsType;
-  }
-
-  @Override
-  public boolean equals(Object obj) {
-    if (obj != null && obj instanceof BeamSqlRowType) {
-      BeamSqlRowType ins = (BeamSqlRowType) obj;
-      return fieldsType.equals(ins.getFieldsType()) && getFieldsName().equals(ins.getFieldsName());
-    } else {
-      return false;
-    }
-  }
-
-  @Override
-  public int hashCode() {
-    return 31 * getFieldsName().hashCode() + getFieldsType().hashCode();
-  }
-}

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlTable.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlTable.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlTable.java
index c179935..b370d9d 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlTable.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamSqlTable.java
@@ -20,6 +20,7 @@ package org.apache.beam.sdk.extensions.sql.schema;
 
 import org.apache.beam.sdk.Pipeline;
 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;
 
@@ -37,16 +38,16 @@ public interface BeamSqlTable {
    * create a {@code PCollection<BeamSqlRow>} from source.
    *
    */
-  PCollection<BeamSqlRow> buildIOReader(Pipeline pipeline);
+  PCollection<BeamRecord> buildIOReader(Pipeline pipeline);
 
   /**
    * create a {@code IO.write()} instance to write to target.
    *
    */
-   PTransform<? super PCollection<BeamSqlRow>, PDone> buildIOWriter();
+   PTransform<? super PCollection<BeamRecord>, PDone> buildIOWriter();
 
   /**
    * Get the schema info of the table.
    */
-   BeamSqlRowType getRowType();
+   BeamSqlRecordType getRowType();
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamTableUtils.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamTableUtils.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamTableUtils.java
index c769928..63c9720 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamTableUtils.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/BeamTableUtils.java
@@ -23,6 +23,7 @@ import java.io.StringReader;
 import java.io.StringWriter;
 import java.math.BigDecimal;
 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;
@@ -34,11 +35,11 @@ import org.apache.commons.csv.CSVRecord;
  * Utility methods for working with {@code BeamTable}.
  */
 public final class BeamTableUtils {
-  public static BeamSqlRow csvLine2BeamSqlRow(
+  public static BeamRecord csvLine2BeamSqlRow(
       CSVFormat csvFormat,
       String line,
-      BeamSqlRowType beamSqlRowType) {
-    BeamSqlRow row = new BeamSqlRow(beamSqlRowType);
+      BeamSqlRecordType beamSqlRowType) {
+    BeamRecord row = new BeamRecord(beamSqlRowType);
     try (StringReader reader = new StringReader(line)) {
       CSVParser parser = csvFormat.parse(reader);
       CSVRecord rawRecord = parser.getRecords().get(0);
@@ -60,7 +61,7 @@ public final class BeamTableUtils {
     return row;
   }
 
-  public static String beamSqlRow2CsvLine(BeamSqlRow row, CSVFormat csvFormat) {
+  public static String beamSqlRow2CsvLine(BeamRecord row, CSVFormat csvFormat) {
     StringWriter writer = new StringWriter();
     try (CSVPrinter printer = csvFormat.print(writer)) {
       for (int i = 0; i < row.size(); i++) {
@@ -73,13 +74,14 @@ public final class BeamTableUtils {
     return writer.toString();
   }
 
-  public static void addFieldWithAutoTypeCasting(BeamSqlRow row, int idx, Object rawObj) {
+  public static void addFieldWithAutoTypeCasting(BeamRecord row, int idx, Object rawObj) {
     if (rawObj == null) {
       row.addField(idx, null);
       return;
     }
 
-    SqlTypeName columnType = CalciteUtils.getFieldType(row.getDataType(), idx);
+    SqlTypeName columnType = CalciteUtils.getFieldType(BeamSqlRecordHelper.getSqlRecordType(row)
+        , idx);
     // auto-casting for numberics
     if ((rawObj instanceof String && SqlTypeName.NUMERIC_TYPES.contains(columnType))
         || (rawObj instanceof BigDecimal && columnType != SqlTypeName.DECIMAL)) {

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/kafka/BeamKafkaCSVTable.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/kafka/BeamKafkaCSVTable.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/kafka/BeamKafkaCSVTable.java
index 2a50947..f137379 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/kafka/BeamKafkaCSVTable.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/kafka/BeamKafkaCSVTable.java
@@ -18,12 +18,12 @@
 package org.apache.beam.sdk.extensions.sql.schema.kafka;
 
 import java.util.List;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRow;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRowType;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordType;
 import org.apache.beam.sdk.extensions.sql.schema.BeamTableUtils;
 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;
@@ -34,45 +34,45 @@ import org.apache.commons.csv.CSVFormat;
  */
 public class BeamKafkaCSVTable extends BeamKafkaTable {
   private CSVFormat csvFormat;
-  public BeamKafkaCSVTable(BeamSqlRowType beamSqlRowType, String bootstrapServers,
+  public BeamKafkaCSVTable(BeamSqlRecordType beamSqlRowType, String bootstrapServers,
       List<String> topics) {
     this(beamSqlRowType, bootstrapServers, topics, CSVFormat.DEFAULT);
   }
 
-  public BeamKafkaCSVTable(BeamSqlRowType beamSqlRowType, String bootstrapServers,
+  public BeamKafkaCSVTable(BeamSqlRecordType beamSqlRowType, String bootstrapServers,
       List<String> topics, CSVFormat format) {
     super(beamSqlRowType, bootstrapServers, topics);
     this.csvFormat = format;
   }
 
   @Override
-  public PTransform<PCollection<KV<byte[], byte[]>>, PCollection<BeamSqlRow>>
+  public PTransform<PCollection<KV<byte[], byte[]>>, PCollection<BeamRecord>>
       getPTransformForInput() {
     return new CsvRecorderDecoder(beamSqlRowType, csvFormat);
   }
 
   @Override
-  public PTransform<PCollection<BeamSqlRow>, PCollection<KV<byte[], byte[]>>>
+  public PTransform<PCollection<BeamRecord>, PCollection<KV<byte[], byte[]>>>
       getPTransformForOutput() {
     return new CsvRecorderEncoder(beamSqlRowType, csvFormat);
   }
 
   /**
-   * A PTransform to convert {@code KV<byte[], byte[]>} to {@link BeamSqlRow}.
+   * A PTransform to convert {@code KV<byte[], byte[]>} to {@link BeamRecord}.
    *
    */
   public static class CsvRecorderDecoder
-      extends PTransform<PCollection<KV<byte[], byte[]>>, PCollection<BeamSqlRow>> {
-    private BeamSqlRowType rowType;
+      extends PTransform<PCollection<KV<byte[], byte[]>>, PCollection<BeamRecord>> {
+    private BeamSqlRecordType rowType;
     private CSVFormat format;
-    public CsvRecorderDecoder(BeamSqlRowType rowType, CSVFormat format) {
+    public CsvRecorderDecoder(BeamSqlRecordType rowType, CSVFormat format) {
       this.rowType = rowType;
       this.format = format;
     }
 
     @Override
-    public PCollection<BeamSqlRow> expand(PCollection<KV<byte[], byte[]>> input) {
-      return input.apply("decodeRecord", ParDo.of(new DoFn<KV<byte[], byte[]>, BeamSqlRow>() {
+    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());
@@ -83,24 +83,24 @@ public class BeamKafkaCSVTable extends BeamKafkaTable {
   }
 
   /**
-   * A PTransform to convert {@link BeamSqlRow} to {@code KV<byte[], byte[]>}.
+   * A PTransform to convert {@link BeamRecord} to {@code KV<byte[], byte[]>}.
    *
    */
   public static class CsvRecorderEncoder
-      extends PTransform<PCollection<BeamSqlRow>, PCollection<KV<byte[], byte[]>>> {
-    private BeamSqlRowType rowType;
+      extends PTransform<PCollection<BeamRecord>, PCollection<KV<byte[], byte[]>>> {
+    private BeamSqlRecordType rowType;
     private CSVFormat format;
-    public CsvRecorderEncoder(BeamSqlRowType rowType, CSVFormat format) {
+    public CsvRecorderEncoder(BeamSqlRecordType rowType, CSVFormat format) {
       this.rowType = rowType;
       this.format = format;
     }
 
     @Override
-    public PCollection<KV<byte[], byte[]>> expand(PCollection<BeamSqlRow> input) {
-      return input.apply("encodeRecord", ParDo.of(new DoFn<BeamSqlRow, KV<byte[], byte[]>>() {
+    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) {
-          BeamSqlRow in = c.element();
+          BeamRecord in = c.element();
           c.output(KV.of(new byte[] {}, BeamTableUtils.beamSqlRow2CsvLine(in, format).getBytes()));
         }
       }));

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/kafka/BeamKafkaTable.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/kafka/BeamKafkaTable.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/kafka/BeamKafkaTable.java
index 2cc664f..fac57bf 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/kafka/BeamKafkaTable.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/kafka/BeamKafkaTable.java
@@ -26,10 +26,10 @@ import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.coders.ByteArrayCoder;
 import org.apache.beam.sdk.extensions.sql.schema.BaseBeamTable;
 import org.apache.beam.sdk.extensions.sql.schema.BeamIOType;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRow;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRowType;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordType;
 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;
@@ -48,11 +48,11 @@ public abstract class BeamKafkaTable extends BaseBeamTable implements Serializab
   private List<String> topics;
   private Map<String, Object> configUpdates;
 
-  protected BeamKafkaTable(BeamSqlRowType beamSqlRowType) {
+  protected BeamKafkaTable(BeamSqlRecordType beamSqlRowType) {
     super(beamSqlRowType);
   }
 
-  public BeamKafkaTable(BeamSqlRowType beamSqlRowType, String bootstrapServers,
+  public BeamKafkaTable(BeamSqlRecordType beamSqlRowType, String bootstrapServers,
       List<String> topics) {
     super(beamSqlRowType);
     this.bootstrapServers = bootstrapServers;
@@ -69,14 +69,14 @@ public abstract class BeamKafkaTable extends BaseBeamTable implements Serializab
     return BeamIOType.UNBOUNDED;
   }
 
-  public abstract PTransform<PCollection<KV<byte[], byte[]>>, PCollection<BeamSqlRow>>
+  public abstract PTransform<PCollection<KV<byte[], byte[]>>, PCollection<BeamRecord>>
       getPTransformForInput();
 
-  public abstract PTransform<PCollection<BeamSqlRow>, PCollection<KV<byte[], byte[]>>>
+  public abstract PTransform<PCollection<BeamRecord>, PCollection<KV<byte[], byte[]>>>
       getPTransformForOutput();
 
   @Override
-  public PCollection<BeamSqlRow> buildIOReader(Pipeline pipeline) {
+  public PCollection<BeamRecord> buildIOReader(Pipeline pipeline) {
     return PBegin.in(pipeline).apply("read",
             KafkaIO.<byte[], byte[]>read()
                 .withBootstrapServers(bootstrapServers)
@@ -89,13 +89,13 @@ public abstract class BeamKafkaTable extends BaseBeamTable implements Serializab
   }
 
   @Override
-  public PTransform<? super PCollection<BeamSqlRow>, PDone> buildIOWriter() {
+  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<BeamSqlRow>, PDone>() {
+    return new PTransform<PCollection<BeamRecord>, PDone>() {
       @Override
-      public PDone expand(PCollection<BeamSqlRow> input) {
+      public PDone expand(PCollection<BeamRecord> input) {
         return input.apply("out_reformat", getPTransformForOutput()).apply("persistent",
             KafkaIO.<byte[], byte[]>write()
                 .withBootstrapServers(bootstrapServers)

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTable.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTable.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTable.java
index c44faab..0ec418c 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTable.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTable.java
@@ -19,10 +19,10 @@
 package org.apache.beam.sdk.extensions.sql.schema.text;
 
 import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRow;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRowType;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordType;
 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;
@@ -46,25 +46,25 @@ public class BeamTextCSVTable extends BeamTextTable {
   /**
    * CSV table with {@link CSVFormat#DEFAULT DEFAULT} format.
    */
-  public BeamTextCSVTable(BeamSqlRowType beamSqlRowType, String filePattern)  {
+  public BeamTextCSVTable(BeamSqlRecordType beamSqlRowType, String filePattern)  {
     this(beamSqlRowType, filePattern, CSVFormat.DEFAULT);
   }
 
-  public BeamTextCSVTable(BeamSqlRowType beamSqlRowType, String filePattern,
+  public BeamTextCSVTable(BeamSqlRecordType beamSqlRowType, String filePattern,
       CSVFormat csvFormat) {
     super(beamSqlRowType, filePattern);
     this.csvFormat = csvFormat;
   }
 
   @Override
-  public PCollection<BeamSqlRow> buildIOReader(Pipeline pipeline) {
+  public PCollection<BeamRecord> buildIOReader(Pipeline pipeline) {
     return PBegin.in(pipeline).apply("decodeRecord", TextIO.read().from(filePattern))
         .apply("parseCSVLine",
             new BeamTextCSVTableIOReader(beamSqlRowType, filePattern, csvFormat));
   }
 
   @Override
-  public PTransform<? super PCollection<BeamSqlRow>, PDone> buildIOWriter() {
+  public PTransform<? super PCollection<BeamRecord>, PDone> buildIOWriter() {
     return new BeamTextCSVTableIOWriter(beamSqlRowType, filePattern, csvFormat);
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTableIOReader.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTableIOReader.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTableIOReader.java
index 06109c3..ecb77e0 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTableIOReader.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTableIOReader.java
@@ -19,12 +19,12 @@
 package org.apache.beam.sdk.extensions.sql.schema.text;
 
 import java.io.Serializable;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRow;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRowType;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordType;
 import org.apache.beam.sdk.extensions.sql.schema.BeamTableUtils;
 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;
 
@@ -32,13 +32,13 @@ import org.apache.commons.csv.CSVFormat;
  * IOReader for {@code BeamTextCSVTable}.
  */
 public class BeamTextCSVTableIOReader
-    extends PTransform<PCollection<String>, PCollection<BeamSqlRow>>
+    extends PTransform<PCollection<String>, PCollection<BeamRecord>>
     implements Serializable {
   private String filePattern;
-  protected BeamSqlRowType beamSqlRowType;
+  protected BeamSqlRecordType beamSqlRowType;
   protected CSVFormat csvFormat;
 
-  public BeamTextCSVTableIOReader(BeamSqlRowType beamSqlRowType, String filePattern,
+  public BeamTextCSVTableIOReader(BeamSqlRecordType beamSqlRowType, String filePattern,
       CSVFormat csvFormat) {
     this.filePattern = filePattern;
     this.beamSqlRowType = beamSqlRowType;
@@ -46,8 +46,8 @@ public class BeamTextCSVTableIOReader
   }
 
   @Override
-  public PCollection<BeamSqlRow> expand(PCollection<String> input) {
-    return input.apply(ParDo.of(new DoFn<String, BeamSqlRow>() {
+  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();

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTableIOWriter.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTableIOWriter.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTableIOWriter.java
index 1684b37..c616973 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTableIOWriter.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextCSVTableIOWriter.java
@@ -19,13 +19,13 @@
 package org.apache.beam.sdk.extensions.sql.schema.text;
 
 import java.io.Serializable;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRow;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRowType;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordType;
 import org.apache.beam.sdk.extensions.sql.schema.BeamTableUtils;
 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;
@@ -33,24 +33,24 @@ import org.apache.commons.csv.CSVFormat;
 /**
  * IOWriter for {@code BeamTextCSVTable}.
  */
-public class BeamTextCSVTableIOWriter extends PTransform<PCollection<BeamSqlRow>, PDone>
+public class BeamTextCSVTableIOWriter extends PTransform<PCollection<BeamRecord>, PDone>
     implements Serializable {
   private String filePattern;
-  protected BeamSqlRowType beamSqlRowType;
+  protected BeamSqlRecordType beamSqlRowType;
   protected CSVFormat csvFormat;
 
-  public BeamTextCSVTableIOWriter(BeamSqlRowType beamSqlRowType, String filePattern,
+  public BeamTextCSVTableIOWriter(BeamSqlRecordType beamSqlRowType, String filePattern,
       CSVFormat csvFormat) {
     this.filePattern = filePattern;
     this.beamSqlRowType = beamSqlRowType;
     this.csvFormat = csvFormat;
   }
 
-  @Override public PDone expand(PCollection<BeamSqlRow> input) {
-    return input.apply("encodeRecord", ParDo.of(new DoFn<BeamSqlRow, String>() {
+  @Override public PDone expand(PCollection<BeamRecord> input) {
+    return input.apply("encodeRecord", ParDo.of(new DoFn<BeamRecord, String>() {
 
       @ProcessElement public void processElement(ProcessContext ctx) {
-        BeamSqlRow row = ctx.element();
+        BeamRecord row = ctx.element();
         ctx.output(BeamTableUtils.beamSqlRow2CsvLine(row, csvFormat));
       }
     })).apply(TextIO.write().to(filePattern));

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextTable.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextTable.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextTable.java
index e85608d..4284366 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextTable.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/schema/text/BeamTextTable.java
@@ -21,7 +21,7 @@ package org.apache.beam.sdk.extensions.sql.schema.text;
 import java.io.Serializable;
 import org.apache.beam.sdk.extensions.sql.schema.BaseBeamTable;
 import org.apache.beam.sdk.extensions.sql.schema.BeamIOType;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRowType;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordType;
 
 /**
  * {@code BeamTextTable} represents a text file/directory(backed by {@code TextIO}).
@@ -29,7 +29,7 @@ import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRowType;
 public abstract class BeamTextTable extends BaseBeamTable implements Serializable {
   protected String filePattern;
 
-  protected BeamTextTable(BeamSqlRowType beamSqlRowType, String filePattern) {
+  protected BeamTextTable(BeamSqlRecordType beamSqlRowType, String filePattern) {
     super(beamSqlRowType);
     this.filePattern = filePattern;
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslAggregationTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslAggregationTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslAggregationTest.java
index e6ca18f..8501157 100644
--- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslAggregationTest.java
+++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslAggregationTest.java
@@ -19,9 +19,9 @@ package org.apache.beam.sdk.extensions.sql;
 
 import java.sql.Types;
 import java.util.Arrays;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRow;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRowType;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordType;
 import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.values.BeamRecord;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionTuple;
 import org.apache.beam.sdk.values.TupleTag;
@@ -49,16 +49,16 @@ public class BeamSqlDslAggregationTest extends BeamSqlDslBase {
     runAggregationWithoutWindow(unboundedInput1);
   }
 
-  private void runAggregationWithoutWindow(PCollection<BeamSqlRow> input) throws Exception {
+  private void runAggregationWithoutWindow(PCollection<BeamRecord> input) throws Exception {
     String sql = "SELECT f_int2, COUNT(*) AS `size` FROM PCOLLECTION GROUP BY f_int2";
 
-    PCollection<BeamSqlRow> result =
+    PCollection<BeamRecord> result =
         input.apply("testAggregationWithoutWindow", BeamSql.simpleQuery(sql));
 
-    BeamSqlRowType resultType = BeamSqlRowType.create(Arrays.asList("f_int2", "size"),
+    BeamSqlRecordType resultType = BeamSqlRecordType.create(Arrays.asList("f_int2", "size"),
         Arrays.asList(Types.INTEGER, Types.BIGINT));
 
-    BeamSqlRow record = new BeamSqlRow(resultType);
+    BeamRecord record = new BeamRecord(resultType);
     record.addField("f_int2", 0);
     record.addField("size", 4L);
 
@@ -83,7 +83,7 @@ public class BeamSqlDslAggregationTest extends BeamSqlDslBase {
     runAggregationFunctions(unboundedInput1);
   }
 
-  private void runAggregationFunctions(PCollection<BeamSqlRow> input) throws Exception{
+  private void runAggregationFunctions(PCollection<BeamRecord> input) throws Exception{
     String sql = "select f_int2, count(*) as size, "
         + "sum(f_long) as sum1, avg(f_long) as avg1, max(f_long) as max1, min(f_long) as min1,"
         + "sum(f_short) as sum2, avg(f_short) as avg2, max(f_short) as max2, min(f_short) as min2,"
@@ -94,11 +94,11 @@ public class BeamSqlDslAggregationTest extends BeamSqlDslBase {
         + "max(f_timestamp) as max6, min(f_timestamp) as min6 "
         + "FROM TABLE_A group by f_int2";
 
-    PCollection<BeamSqlRow> result =
-        PCollectionTuple.of(new TupleTag<BeamSqlRow>("TABLE_A"), input)
+    PCollection<BeamRecord> result =
+        PCollectionTuple.of(new TupleTag<BeamRecord>("TABLE_A"), input)
         .apply("testAggregationFunctions", BeamSql.query(sql));
 
-    BeamSqlRowType resultType = BeamSqlRowType.create(
+    BeamSqlRecordType resultType = BeamSqlRecordType.create(
         Arrays.asList("f_int2", "size", "sum1", "avg1", "max1", "min1", "sum2", "avg2", "max2",
             "min2", "sum3", "avg3", "max3", "min3", "sum4", "avg4", "max4", "min4", "sum5", "avg5",
             "max5", "min5", "max6", "min6"),
@@ -108,7 +108,7 @@ public class BeamSqlDslAggregationTest extends BeamSqlDslBase {
             Types.FLOAT, Types.FLOAT, Types.DOUBLE, Types.DOUBLE, Types.DOUBLE, Types.DOUBLE,
             Types.TIMESTAMP, Types.TIMESTAMP));
 
-    BeamSqlRow record = new BeamSqlRow(resultType);
+    BeamRecord record = new BeamRecord(resultType);
     record.addField("f_int2", 0);
     record.addField("size", 4L);
 
@@ -161,28 +161,28 @@ public class BeamSqlDslAggregationTest extends BeamSqlDslBase {
     runDistinct(unboundedInput1);
   }
 
-  private void runDistinct(PCollection<BeamSqlRow> input) throws Exception {
+  private void runDistinct(PCollection<BeamRecord> input) throws Exception {
     String sql = "SELECT distinct f_int, f_long FROM PCOLLECTION ";
 
-    PCollection<BeamSqlRow> result =
+    PCollection<BeamRecord> result =
         input.apply("testDistinct", BeamSql.simpleQuery(sql));
 
-    BeamSqlRowType resultType = BeamSqlRowType.create(Arrays.asList("f_int", "f_long"),
+    BeamSqlRecordType resultType = BeamSqlRecordType.create(Arrays.asList("f_int", "f_long"),
         Arrays.asList(Types.INTEGER, Types.BIGINT));
 
-    BeamSqlRow record1 = new BeamSqlRow(resultType);
+    BeamRecord record1 = new BeamRecord(resultType);
     record1.addField("f_int", 1);
     record1.addField("f_long", 1000L);
 
-    BeamSqlRow record2 = new BeamSqlRow(resultType);
+    BeamRecord record2 = new BeamRecord(resultType);
     record2.addField("f_int", 2);
     record2.addField("f_long", 2000L);
 
-    BeamSqlRow record3 = new BeamSqlRow(resultType);
+    BeamRecord record3 = new BeamRecord(resultType);
     record3.addField("f_int", 3);
     record3.addField("f_long", 3000L);
 
-    BeamSqlRow record4 = new BeamSqlRow(resultType);
+    BeamRecord record4 = new BeamRecord(resultType);
     record4.addField("f_int", 4);
     record4.addField("f_long", 4000L);
 
@@ -207,27 +207,27 @@ public class BeamSqlDslAggregationTest extends BeamSqlDslBase {
     runTumbleWindow(unboundedInput1);
   }
 
-  private void runTumbleWindow(PCollection<BeamSqlRow> input) throws Exception {
+  private void runTumbleWindow(PCollection<BeamRecord> input) throws Exception {
     String sql = "SELECT f_int2, COUNT(*) AS `size`,"
         + " TUMBLE_START(f_timestamp, INTERVAL '1' HOUR) AS `window_start`"
         + " FROM TABLE_A"
         + " GROUP BY f_int2, TUMBLE(f_timestamp, INTERVAL '1' HOUR)";
-    PCollection<BeamSqlRow> result =
-        PCollectionTuple.of(new TupleTag<BeamSqlRow>("TABLE_A"), input)
+    PCollection<BeamRecord> result =
+        PCollectionTuple.of(new TupleTag<BeamRecord>("TABLE_A"), input)
         .apply("testTumbleWindow", BeamSql.query(sql));
 
-    BeamSqlRowType resultType = BeamSqlRowType.create(
+    BeamSqlRecordType resultType = BeamSqlRecordType.create(
         Arrays.asList("f_int2", "size", "window_start"),
         Arrays.asList(Types.INTEGER, Types.BIGINT, Types.TIMESTAMP));
 
-    BeamSqlRow record1 = new BeamSqlRow(resultType);
+    BeamRecord record1 = new BeamRecord(resultType);
     record1.addField("f_int2", 0);
     record1.addField("size", 3L);
     record1.addField("window_start", FORMAT.parse("2017-01-01 01:00:00"));
     record1.setWindowStart(new Instant(FORMAT.parse("2017-01-01 01:00:00").getTime()));
     record1.setWindowEnd(new Instant(FORMAT.parse("2017-01-01 02:00:00").getTime()));
 
-    BeamSqlRow record2 = new BeamSqlRow(resultType);
+    BeamRecord record2 = new BeamRecord(resultType);
     record2.addField("f_int2", 0);
     record2.addField("size", 1L);
     record2.addField("window_start", FORMAT.parse("2017-01-01 02:00:00"));
@@ -255,40 +255,40 @@ public class BeamSqlDslAggregationTest extends BeamSqlDslBase {
     runHopWindow(unboundedInput1);
   }
 
-  private void runHopWindow(PCollection<BeamSqlRow> input) throws Exception {
+  private void runHopWindow(PCollection<BeamRecord> input) throws Exception {
     String sql = "SELECT f_int2, COUNT(*) AS `size`,"
         + " HOP_START(f_timestamp, INTERVAL '1' HOUR, INTERVAL '30' MINUTE) AS `window_start`"
         + " FROM PCOLLECTION"
         + " GROUP BY f_int2, HOP(f_timestamp, INTERVAL '1' HOUR, INTERVAL '30' MINUTE)";
-    PCollection<BeamSqlRow> result =
+    PCollection<BeamRecord> result =
         input.apply("testHopWindow", BeamSql.simpleQuery(sql));
 
-    BeamSqlRowType resultType = BeamSqlRowType.create(
+    BeamSqlRecordType resultType = BeamSqlRecordType.create(
         Arrays.asList("f_int2", "size", "window_start"),
         Arrays.asList(Types.INTEGER, Types.BIGINT, Types.TIMESTAMP));
 
-    BeamSqlRow record1 = new BeamSqlRow(resultType);
+    BeamRecord record1 = new BeamRecord(resultType);
     record1.addField("f_int2", 0);
     record1.addField("size", 3L);
     record1.addField("window_start", FORMAT.parse("2017-01-01 00:30:00"));
     record1.setWindowStart(new Instant(FORMAT.parse("2017-01-01 00:30:00").getTime()));
     record1.setWindowEnd(new Instant(FORMAT.parse("2017-01-01 01:30:00").getTime()));
 
-    BeamSqlRow record2 = new BeamSqlRow(resultType);
+    BeamRecord record2 = new BeamRecord(resultType);
     record2.addField("f_int2", 0);
     record2.addField("size", 3L);
     record2.addField("window_start", FORMAT.parse("2017-01-01 01:00:00"));
     record2.setWindowStart(new Instant(FORMAT.parse("2017-01-01 01:00:00").getTime()));
     record2.setWindowEnd(new Instant(FORMAT.parse("2017-01-01 02:00:00").getTime()));
 
-    BeamSqlRow record3 = new BeamSqlRow(resultType);
+    BeamRecord record3 = new BeamRecord(resultType);
     record3.addField("f_int2", 0);
     record3.addField("size", 1L);
     record3.addField("window_start", FORMAT.parse("2017-01-01 01:30:00"));
     record3.setWindowStart(new Instant(FORMAT.parse("2017-01-01 01:30:00").getTime()));
     record3.setWindowEnd(new Instant(FORMAT.parse("2017-01-01 02:30:00").getTime()));
 
-    BeamSqlRow record4 = new BeamSqlRow(resultType);
+    BeamRecord record4 = new BeamRecord(resultType);
     record4.addField("f_int2", 0);
     record4.addField("size", 1L);
     record4.addField("window_start", FORMAT.parse("2017-01-01 02:00:00"));
@@ -316,27 +316,27 @@ public class BeamSqlDslAggregationTest extends BeamSqlDslBase {
     runSessionWindow(unboundedInput1);
   }
 
-  private void runSessionWindow(PCollection<BeamSqlRow> input) throws Exception {
+  private void runSessionWindow(PCollection<BeamRecord> input) throws Exception {
     String sql = "SELECT f_int2, COUNT(*) AS `size`,"
         + " SESSION_START(f_timestamp, INTERVAL '5' MINUTE) AS `window_start`"
         + " FROM TABLE_A"
         + " GROUP BY f_int2, SESSION(f_timestamp, INTERVAL '5' MINUTE)";
-    PCollection<BeamSqlRow> result =
-        PCollectionTuple.of(new TupleTag<BeamSqlRow>("TABLE_A"), input)
+    PCollection<BeamRecord> result =
+        PCollectionTuple.of(new TupleTag<BeamRecord>("TABLE_A"), input)
         .apply("testSessionWindow", BeamSql.query(sql));
 
-    BeamSqlRowType resultType = BeamSqlRowType.create(
+    BeamSqlRecordType resultType = BeamSqlRecordType.create(
         Arrays.asList("f_int2", "size", "window_start"),
         Arrays.asList(Types.INTEGER, Types.BIGINT, Types.TIMESTAMP));
 
-    BeamSqlRow record1 = new BeamSqlRow(resultType);
+    BeamRecord record1 = new BeamRecord(resultType);
     record1.addField("f_int2", 0);
     record1.addField("size", 3L);
     record1.addField("window_start", FORMAT.parse("2017-01-01 01:01:03"));
     record1.setWindowStart(new Instant(FORMAT.parse("2017-01-01 01:01:03").getTime()));
     record1.setWindowEnd(new Instant(FORMAT.parse("2017-01-01 01:11:03").getTime()));
 
-    BeamSqlRow record2 = new BeamSqlRow(resultType);
+    BeamRecord record2 = new BeamRecord(resultType);
     record2.addField("f_int2", 0);
     record2.addField("size", 1L);
     record2.addField("window_start", FORMAT.parse("2017-01-01 02:04:03"));
@@ -357,8 +357,8 @@ public class BeamSqlDslAggregationTest extends BeamSqlDslBase {
 
     String sql = "SELECT f_int2, COUNT(*) AS `size` FROM TABLE_A "
         + "GROUP BY f_int2, TUMBLE(f_long, INTERVAL '1' HOUR)";
-    PCollection<BeamSqlRow> result =
-        PCollectionTuple.of(new TupleTag<BeamSqlRow>("TABLE_A"), boundedInput1)
+    PCollection<BeamRecord> result =
+        PCollectionTuple.of(new TupleTag<BeamRecord>("TABLE_A"), boundedInput1)
         .apply("testWindowOnNonTimestampField", BeamSql.query(sql));
 
     pipeline.run().waitUntilFinish();
@@ -372,7 +372,7 @@ public class BeamSqlDslAggregationTest extends BeamSqlDslBase {
 
     String sql = "SELECT f_int2, COUNT(DISTINCT *) AS `size` FROM PCOLLECTION GROUP BY f_int2";
 
-    PCollection<BeamSqlRow> result =
+    PCollection<BeamRecord> result =
         boundedInput1.apply("testUnsupportedDistinct", BeamSql.simpleQuery(sql));
 
     pipeline.run().waitUntilFinish();

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslBase.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslBase.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslBase.java
index 0c1ce1c..d09caf0 100644
--- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslBase.java
+++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslBase.java
@@ -25,12 +25,11 @@ import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRow;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRowCoder;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRowType;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordType;
 import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.testing.TestStream;
 import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.BeamRecord;
 import org.apache.beam.sdk.values.PBegin;
 import org.apache.beam.sdk.values.PCollection;
 import org.joda.time.Instant;
@@ -53,20 +52,20 @@ public class BeamSqlDslBase {
   @Rule
   public ExpectedException exceptions = ExpectedException.none();
 
-  public static BeamSqlRowType rowTypeInTableA;
-  public static List<BeamSqlRow> recordsInTableA;
+  public static BeamSqlRecordType rowTypeInTableA;
+  public static List<BeamRecord> recordsInTableA;
 
   //bounded PCollections
-  public PCollection<BeamSqlRow> boundedInput1;
-  public PCollection<BeamSqlRow> boundedInput2;
+  public PCollection<BeamRecord> boundedInput1;
+  public PCollection<BeamRecord> boundedInput2;
 
   //unbounded PCollections
-  public PCollection<BeamSqlRow> unboundedInput1;
-  public PCollection<BeamSqlRow> unboundedInput2;
+  public PCollection<BeamRecord> unboundedInput1;
+  public PCollection<BeamRecord> unboundedInput2;
 
   @BeforeClass
   public static void prepareClass() throws ParseException {
-    rowTypeInTableA = BeamSqlRowType.create(
+    rowTypeInTableA = BeamSqlRecordType.create(
         Arrays.asList("f_int", "f_long", "f_short", "f_byte", "f_float", "f_double", "f_string",
             "f_timestamp", "f_int2", "f_decimal"),
         Arrays.asList(Types.INTEGER, Types.BIGINT, Types.SMALLINT, Types.TINYINT, Types.FLOAT,
@@ -78,20 +77,20 @@ public class BeamSqlDslBase {
   @Before
   public void preparePCollections(){
     boundedInput1 = PBegin.in(pipeline).apply("boundedInput1",
-        Create.of(recordsInTableA).withCoder(new BeamSqlRowCoder(rowTypeInTableA)));
+        Create.of(recordsInTableA).withCoder(rowTypeInTableA.getRecordCoder()));
 
     boundedInput2 = PBegin.in(pipeline).apply("boundedInput2",
-        Create.of(recordsInTableA.get(0)).withCoder(new BeamSqlRowCoder(rowTypeInTableA)));
+        Create.of(recordsInTableA.get(0)).withCoder(rowTypeInTableA.getRecordCoder()));
 
     unboundedInput1 = prepareUnboundedPCollection1();
     unboundedInput2 = prepareUnboundedPCollection2();
   }
 
-  private PCollection<BeamSqlRow> prepareUnboundedPCollection1() {
-    TestStream.Builder<BeamSqlRow> values = TestStream
-        .create(new BeamSqlRowCoder(rowTypeInTableA));
+  private PCollection<BeamRecord> prepareUnboundedPCollection1() {
+    TestStream.Builder<BeamRecord> values = TestStream
+        .create(rowTypeInTableA.getRecordCoder());
 
-    for (BeamSqlRow row : recordsInTableA) {
+    for (BeamRecord row : recordsInTableA) {
       values = values.advanceWatermarkTo(new Instant(row.getDate("f_timestamp")));
       values = values.addElements(row);
     }
@@ -99,21 +98,21 @@ public class BeamSqlDslBase {
     return PBegin.in(pipeline).apply("unboundedInput1", values.advanceWatermarkToInfinity());
   }
 
-  private PCollection<BeamSqlRow> prepareUnboundedPCollection2() {
-    TestStream.Builder<BeamSqlRow> values = TestStream
-        .create(new BeamSqlRowCoder(rowTypeInTableA));
+  private PCollection<BeamRecord> prepareUnboundedPCollection2() {
+    TestStream.Builder<BeamRecord> values = TestStream
+        .create(rowTypeInTableA.getRecordCoder());
 
-    BeamSqlRow row = recordsInTableA.get(0);
+    BeamRecord row = recordsInTableA.get(0);
     values = values.advanceWatermarkTo(new Instant(row.getDate("f_timestamp")));
     values = values.addElements(row);
 
     return PBegin.in(pipeline).apply("unboundedInput2", values.advanceWatermarkToInfinity());
   }
 
-  private static List<BeamSqlRow> prepareInputRowsInTableA() throws ParseException{
-    List<BeamSqlRow> rows = new ArrayList<>();
+  private static List<BeamRecord> prepareInputRowsInTableA() throws ParseException{
+    List<BeamRecord> rows = new ArrayList<>();
 
-    BeamSqlRow row1 = new BeamSqlRow(rowTypeInTableA);
+    BeamRecord row1 = new BeamRecord(rowTypeInTableA);
     row1.addField(0, 1);
     row1.addField(1, 1000L);
     row1.addField(2, Short.valueOf("1"));
@@ -126,7 +125,7 @@ public class BeamSqlDslBase {
     row1.addField(9, new BigDecimal(1));
     rows.add(row1);
 
-    BeamSqlRow row2 = new BeamSqlRow(rowTypeInTableA);
+    BeamRecord row2 = new BeamRecord(rowTypeInTableA);
     row2.addField(0, 2);
     row2.addField(1, 2000L);
     row2.addField(2, Short.valueOf("2"));
@@ -139,7 +138,7 @@ public class BeamSqlDslBase {
     row2.addField(9, new BigDecimal(2));
     rows.add(row2);
 
-    BeamSqlRow row3 = new BeamSqlRow(rowTypeInTableA);
+    BeamRecord row3 = new BeamRecord(rowTypeInTableA);
     row3.addField(0, 3);
     row3.addField(1, 3000L);
     row3.addField(2, Short.valueOf("3"));
@@ -152,7 +151,7 @@ public class BeamSqlDslBase {
     row3.addField(9, new BigDecimal(3));
     rows.add(row3);
 
-    BeamSqlRow row4 = new BeamSqlRow(rowTypeInTableA);
+    BeamRecord row4 = new BeamRecord(rowTypeInTableA);
     row4.addField(0, 4);
     row4.addField(1, 4000L);
     row4.addField(2, Short.valueOf("4"));

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslFilterTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslFilterTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslFilterTest.java
index 16b6426..e1d463b 100644
--- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslFilterTest.java
+++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslFilterTest.java
@@ -17,8 +17,8 @@
  */
 package org.apache.beam.sdk.extensions.sql;
 
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRow;
 import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.values.BeamRecord;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionTuple;
 import org.apache.beam.sdk.values.TupleTag;
@@ -44,10 +44,10 @@ public class BeamSqlDslFilterTest extends BeamSqlDslBase {
     runSingleFilter(unboundedInput1);
   }
 
-  private void runSingleFilter(PCollection<BeamSqlRow> input) throws Exception {
+  private void runSingleFilter(PCollection<BeamRecord> input) throws Exception {
     String sql = "SELECT * FROM PCOLLECTION WHERE f_int = 1";
 
-    PCollection<BeamSqlRow> result =
+    PCollection<BeamRecord> result =
         input.apply("testSingleFilter", BeamSql.simpleQuery(sql));
 
     PAssert.that(result).containsInAnyOrder(recordsInTableA.get(0));
@@ -71,12 +71,12 @@ public class BeamSqlDslFilterTest extends BeamSqlDslBase {
     runCompositeFilter(unboundedInput1);
   }
 
-  private void runCompositeFilter(PCollection<BeamSqlRow> input) throws Exception {
+  private void runCompositeFilter(PCollection<BeamRecord> input) throws Exception {
     String sql = "SELECT * FROM TABLE_A"
         + " WHERE f_int > 1 AND (f_long < 3000 OR f_string = 'string_row3')";
 
-    PCollection<BeamSqlRow> result =
-        PCollectionTuple.of(new TupleTag<BeamSqlRow>("TABLE_A"), input)
+    PCollection<BeamRecord> result =
+        PCollectionTuple.of(new TupleTag<BeamRecord>("TABLE_A"), input)
         .apply("testCompositeFilter", BeamSql.query(sql));
 
     PAssert.that(result).containsInAnyOrder(recordsInTableA.get(1), recordsInTableA.get(2));
@@ -100,11 +100,11 @@ public class BeamSqlDslFilterTest extends BeamSqlDslBase {
     runNoReturnFilter(unboundedInput1);
   }
 
-  private void runNoReturnFilter(PCollection<BeamSqlRow> input) throws Exception {
+  private void runNoReturnFilter(PCollection<BeamRecord> input) throws Exception {
     String sql = "SELECT * FROM TABLE_A WHERE f_int < 1";
 
-    PCollection<BeamSqlRow> result =
-        PCollectionTuple.of(new TupleTag<BeamSqlRow>("TABLE_A"), input)
+    PCollection<BeamRecord> result =
+        PCollectionTuple.of(new TupleTag<BeamRecord>("TABLE_A"), input)
         .apply("testNoReturnFilter", BeamSql.query(sql));
 
     PAssert.that(result).empty();
@@ -120,8 +120,8 @@ public class BeamSqlDslFilterTest extends BeamSqlDslBase {
 
     String sql = "SELECT * FROM TABLE_B WHERE f_int < 1";
 
-    PCollection<BeamSqlRow> result =
-        PCollectionTuple.of(new TupleTag<BeamSqlRow>("TABLE_A"), boundedInput1)
+    PCollection<BeamRecord> result =
+        PCollectionTuple.of(new TupleTag<BeamRecord>("TABLE_A"), boundedInput1)
         .apply("testFromInvalidTableName1", BeamSql.query(sql));
 
     pipeline.run().waitUntilFinish();
@@ -135,7 +135,7 @@ public class BeamSqlDslFilterTest extends BeamSqlDslBase {
 
     String sql = "SELECT * FROM PCOLLECTION_NA";
 
-    PCollection<BeamSqlRow> result = boundedInput1.apply(BeamSql.simpleQuery(sql));
+    PCollection<BeamRecord> result = boundedInput1.apply(BeamSql.simpleQuery(sql));
 
     pipeline.run().waitUntilFinish();
   }
@@ -148,7 +148,7 @@ public class BeamSqlDslFilterTest extends BeamSqlDslBase {
 
     String sql = "SELECT * FROM PCOLLECTION WHERE f_int_na = 0";
 
-    PCollection<BeamSqlRow> result = boundedInput1.apply(BeamSql.simpleQuery(sql));
+    PCollection<BeamRecord> result = boundedInput1.apply(BeamSql.simpleQuery(sql));
 
     pipeline.run().waitUntilFinish();
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslJoinTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslJoinTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslJoinTest.java
index d75af9b..d5d0a24 100644
--- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslJoinTest.java
+++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslJoinTest.java
@@ -23,11 +23,11 @@ import static org.apache.beam.sdk.extensions.sql.impl.rel.BeamJoinRelBoundedVsBo
 
 import java.sql.Types;
 import java.util.Arrays;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRow;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRowCoder;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRowType;
+import org.apache.beam.sdk.coders.BeamRecordCoder;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordType;
 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.beam.sdk.values.PCollectionTuple;
 import org.apache.beam.sdk.values.TupleTag;
@@ -41,8 +41,8 @@ public class BeamSqlDslJoinTest {
   @Rule
   public final TestPipeline pipeline = TestPipeline.create();
 
-  private static final BeamSqlRowType SOURCE_RECORD_TYPE =
-      BeamSqlRowType.create(
+  private static final BeamSqlRecordType SOURCE_RECORD_TYPE =
+      BeamSqlRecordType.create(
           Arrays.asList(
               "order_id", "site_id", "price"
           ),
@@ -51,11 +51,10 @@ public class BeamSqlDslJoinTest {
           )
       );
 
-  private static final BeamSqlRowCoder SOURCE_CODER =
-      new BeamSqlRowCoder(SOURCE_RECORD_TYPE);
+  private static final BeamRecordCoder SOURCE_CODER = SOURCE_RECORD_TYPE.getRecordCoder();
 
-  private static final BeamSqlRowType RESULT_RECORD_TYPE =
-      BeamSqlRowType.create(
+  private static final BeamSqlRecordType RESULT_RECORD_TYPE =
+      BeamSqlRecordType.create(
           Arrays.asList(
           "order_id", "site_id", "price", "order_id0", "site_id0", "price0"
           ),
@@ -65,8 +64,7 @@ public class BeamSqlDslJoinTest {
           )
       );
 
-  private static final BeamSqlRowCoder RESULT_CODER =
-      new BeamSqlRowCoder(RESULT_RECORD_TYPE);
+  private static final BeamRecordCoder RESULT_CODER = RESULT_RECORD_TYPE.getRecordCoder();
 
   @Test
   public void testInnerJoin() throws Exception {
@@ -178,13 +176,13 @@ public class BeamSqlDslJoinTest {
     pipeline.run();
   }
 
-  private PCollection<BeamSqlRow> queryFromOrderTables(String sql) {
+  private PCollection<BeamRecord> queryFromOrderTables(String sql) {
     return PCollectionTuple
         .of(
-            new TupleTag<BeamSqlRow>("ORDER_DETAILS1"),
+            new TupleTag<BeamRecord>("ORDER_DETAILS1"),
             ORDER_DETAILS1.buildIOReader(pipeline).setCoder(SOURCE_CODER)
         )
-        .and(new TupleTag<BeamSqlRow>("ORDER_DETAILS2"),
+        .and(new TupleTag<BeamRecord>("ORDER_DETAILS2"),
             ORDER_DETAILS2.buildIOReader(pipeline).setCoder(SOURCE_CODER)
         ).apply("join", BeamSql.query(sql)).setCoder(RESULT_CODER);
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/89109b8c/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslProjectTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslProjectTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslProjectTest.java
index 6468011..ddb90d5 100644
--- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslProjectTest.java
+++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslProjectTest.java
@@ -19,9 +19,9 @@ package org.apache.beam.sdk.extensions.sql;
 
 import java.sql.Types;
 import java.util.Arrays;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRow;
-import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRowType;
+import org.apache.beam.sdk.extensions.sql.schema.BeamSqlRecordType;
 import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.values.BeamRecord;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionTuple;
 import org.apache.beam.sdk.values.TupleTag;
@@ -47,10 +47,10 @@ public class BeamSqlDslProjectTest extends BeamSqlDslBase {
     runSelectAll(unboundedInput2);
   }
 
-  private void runSelectAll(PCollection<BeamSqlRow> input) throws Exception {
+  private void runSelectAll(PCollection<BeamRecord> input) throws Exception {
     String sql = "SELECT * FROM PCOLLECTION";
 
-    PCollection<BeamSqlRow> result =
+    PCollection<BeamRecord> result =
         input.apply("testSelectAll", BeamSql.simpleQuery(sql));
 
     PAssert.that(result).containsInAnyOrder(recordsInTableA.get(0));
@@ -74,17 +74,17 @@ public class BeamSqlDslProjectTest extends BeamSqlDslBase {
     runPartialFields(unboundedInput2);
   }
 
-  private void runPartialFields(PCollection<BeamSqlRow> input) throws Exception {
+  private void runPartialFields(PCollection<BeamRecord> input) throws Exception {
     String sql = "SELECT f_int, f_long FROM TABLE_A";
 
-    PCollection<BeamSqlRow> result =
-        PCollectionTuple.of(new TupleTag<BeamSqlRow>("TABLE_A"), input)
+    PCollection<BeamRecord> result =
+        PCollectionTuple.of(new TupleTag<BeamRecord>("TABLE_A"), input)
         .apply("testPartialFields", BeamSql.query(sql));
 
-    BeamSqlRowType resultType = BeamSqlRowType.create(Arrays.asList("f_int", "f_long"),
+    BeamSqlRecordType resultType = BeamSqlRecordType.create(Arrays.asList("f_int", "f_long"),
         Arrays.asList(Types.INTEGER, Types.BIGINT));
 
-    BeamSqlRow record = new BeamSqlRow(resultType);
+    BeamRecord record = new BeamRecord(resultType);
     record.addField("f_int", recordsInTableA.get(0).getFieldValue(0));
     record.addField("f_long", recordsInTableA.get(0).getFieldValue(1));
 
@@ -109,29 +109,29 @@ public class BeamSqlDslProjectTest extends BeamSqlDslBase {
     runPartialFieldsInMultipleRow(unboundedInput1);
   }
 
-  private void runPartialFieldsInMultipleRow(PCollection<BeamSqlRow> input) throws Exception {
+  private void runPartialFieldsInMultipleRow(PCollection<BeamRecord> input) throws Exception {
     String sql = "SELECT f_int, f_long FROM TABLE_A";
 
-    PCollection<BeamSqlRow> result =
-        PCollectionTuple.of(new TupleTag<BeamSqlRow>("TABLE_A"), input)
+    PCollection<BeamRecord> result =
+        PCollectionTuple.of(new TupleTag<BeamRecord>("TABLE_A"), input)
         .apply("testPartialFieldsInMultipleRow", BeamSql.query(sql));
 
-    BeamSqlRowType resultType = BeamSqlRowType.create(Arrays.asList("f_int", "f_long"),
+    BeamSqlRecordType resultType = BeamSqlRecordType.create(Arrays.asList("f_int", "f_long"),
         Arrays.asList(Types.INTEGER, Types.BIGINT));
 
-    BeamSqlRow record1 = new BeamSqlRow(resultType);
+    BeamRecord record1 = new BeamRecord(resultType);
     record1.addField("f_int", recordsInTableA.get(0).getFieldValue(0));
     record1.addField("f_long", recordsInTableA.get(0).getFieldValue(1));
 
-    BeamSqlRow record2 = new BeamSqlRow(resultType);
+    BeamRecord record2 = new BeamRecord(resultType);
     record2.addField("f_int", recordsInTableA.get(1).getFieldValue(0));
     record2.addField("f_long", recordsInTableA.get(1).getFieldValue(1));
 
-    BeamSqlRow record3 = new BeamSqlRow(resultType);
+    BeamRecord record3 = new BeamRecord(resultType);
     record3.addField("f_int", recordsInTableA.get(2).getFieldValue(0));
     record3.addField("f_long", recordsInTableA.get(2).getFieldValue(1));
 
-    BeamSqlRow record4 = new BeamSqlRow(resultType);
+    BeamRecord record4 = new BeamRecord(resultType);
     record4.addField("f_int", recordsInTableA.get(3).getFieldValue(0));
     record4.addField("f_long", recordsInTableA.get(3).getFieldValue(1));
 
@@ -156,29 +156,29 @@ public class BeamSqlDslProjectTest extends BeamSqlDslBase {
     runPartialFieldsInRows(unboundedInput1);
   }
 
-  private void runPartialFieldsInRows(PCollection<BeamSqlRow> input) throws Exception {
+  private void runPartialFieldsInRows(PCollection<BeamRecord> input) throws Exception {
     String sql = "SELECT f_int, f_long FROM TABLE_A";
 
-    PCollection<BeamSqlRow> result =
-        PCollectionTuple.of(new TupleTag<BeamSqlRow>("TABLE_A"), input)
+    PCollection<BeamRecord> result =
+        PCollectionTuple.of(new TupleTag<BeamRecord>("TABLE_A"), input)
         .apply("testPartialFieldsInRows", BeamSql.query(sql));
 
-    BeamSqlRowType resultType = BeamSqlRowType.create(Arrays.asList("f_int", "f_long"),
+    BeamSqlRecordType resultType = BeamSqlRecordType.create(Arrays.asList("f_int", "f_long"),
         Arrays.asList(Types.INTEGER, Types.BIGINT));
 
-    BeamSqlRow record1 = new BeamSqlRow(resultType);
+    BeamRecord record1 = new BeamRecord(resultType);
     record1.addField("f_int", recordsInTableA.get(0).getFieldValue(0));
     record1.addField("f_long", recordsInTableA.get(0).getFieldValue(1));
 
-    BeamSqlRow record2 = new BeamSqlRow(resultType);
+    BeamRecord record2 = new BeamRecord(resultType);
     record2.addField("f_int", recordsInTableA.get(1).getFieldValue(0));
     record2.addField("f_long", recordsInTableA.get(1).getFieldValue(1));
 
-    BeamSqlRow record3 = new BeamSqlRow(resultType);
+    BeamRecord record3 = new BeamRecord(resultType);
     record3.addField("f_int", recordsInTableA.get(2).getFieldValue(0));
     record3.addField("f_long", recordsInTableA.get(2).getFieldValue(1));
 
-    BeamSqlRow record4 = new BeamSqlRow(resultType);
+    BeamRecord record4 = new BeamRecord(resultType);
     record4.addField("f_int", recordsInTableA.get(3).getFieldValue(0));
     record4.addField("f_long", recordsInTableA.get(3).getFieldValue(1));
 
@@ -203,17 +203,17 @@ public class BeamSqlDslProjectTest extends BeamSqlDslBase {
     runLiteralField(unboundedInput2);
   }
 
-  public void runLiteralField(PCollection<BeamSqlRow> input) throws Exception {
+  public void runLiteralField(PCollection<BeamRecord> input) throws Exception {
     String sql = "SELECT 1 as literal_field FROM TABLE_A";
 
-    PCollection<BeamSqlRow> result =
-        PCollectionTuple.of(new TupleTag<BeamSqlRow>("TABLE_A"), input)
+    PCollection<BeamRecord> result =
+        PCollectionTuple.of(new TupleTag<BeamRecord>("TABLE_A"), input)
         .apply("testLiteralField", BeamSql.query(sql));
 
-    BeamSqlRowType resultType = BeamSqlRowType.create(Arrays.asList("literal_field"),
+    BeamSqlRecordType resultType = BeamSqlRecordType.create(Arrays.asList("literal_field"),
         Arrays.asList(Types.INTEGER));
 
-    BeamSqlRow record = new BeamSqlRow(resultType);
+    BeamRecord record = new BeamRecord(resultType);
     record.addField("literal_field", 1);
 
     PAssert.that(result).containsInAnyOrder(record);
@@ -229,8 +229,8 @@ public class BeamSqlDslProjectTest extends BeamSqlDslBase {
 
     String sql = "SELECT f_int_na FROM TABLE_A";
 
-    PCollection<BeamSqlRow> result =
-        PCollectionTuple.of(new TupleTag<BeamSqlRow>("TABLE_A"), boundedInput1)
+    PCollection<BeamRecord> result =
+        PCollectionTuple.of(new TupleTag<BeamRecord>("TABLE_A"), boundedInput1)
         .apply("testProjectUnknownField", BeamSql.query(sql));
 
     pipeline.run().waitUntilFinish();