beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From taki...@apache.org
Subject [1/2] beam git commit: update JavaDoc for BeamRecord, BeamRecordType. Also only create new UDF class instances for SerializableFunction UDFs.
Date Thu, 10 Aug 2017 21:00:35 GMT
Repository: beam
Updated Branches:
  refs/heads/DSL_SQL 2a1377e1c -> f37a7a19c


update JavaDoc for BeamRecord, BeamRecordType.
Also only create new UDF class instances for SerializableFunction UDFs.


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

Branch: refs/heads/DSL_SQL
Commit: cd27036a1c537a9059f13bbcdaec0264481ebd0b
Parents: 2a1377e
Author: mingmxu <mingmxu@ebay.com>
Authored: Wed Aug 9 13:30:16 2017 -0700
Committer: Tyler Akidau <takidau@apache.org>
Committed: Thu Aug 10 13:58:59 2017 -0700

----------------------------------------------------------------------
 .../org/apache/beam/sdk/values/BeamRecord.java  | 118 ++++++++++++++++++-
 .../apache/beam/sdk/values/BeamRecordType.java  |  60 ++++++----
 .../operator/BeamSqlUdfExpression.java          |   5 +-
 3 files changed, 159 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/cd27036a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/BeamRecord.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/BeamRecord.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/BeamRecord.java
index fa3b574..fd26f46 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/BeamRecord.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/BeamRecord.java
@@ -25,11 +25,17 @@ import java.util.Date;
 import java.util.GregorianCalendar;
 import java.util.List;
 import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.coders.BeamRecordCoder;
 
 /**
- * {@link org.apache.beam.sdk.values.BeamRecord}, self-described with
- * {@link BeamRecordType}, represents one element in a
- * {@link org.apache.beam.sdk.values.PCollection}.
+ * {@link BeamRecord} is an immutable tuple-like type to represent one element in a
+ * {@link PCollection}. The fields are described with a {@link BeamRecordType}.
+ *
+ * <p>By default, {@link BeamRecordType} only contains the name for each field. It
+ * can be extended to support more sophisticated validation by overwriting
+ * {@link BeamRecordType#validateValueType(int, Object)}.
+ *
+ * <p>A Coder {@link BeamRecordCoder} is provided, which wraps the Coder for each data
field.
  */
 @Experimental
 public class BeamRecord implements Serializable {
@@ -63,6 +69,9 @@ public class BeamRecord implements Serializable {
     }
   }
 
+  /**
+   * see {@link #BeamRecord(BeamRecordType, List)}.
+   */
   public BeamRecord(BeamRecordType dataType, Object... rawdataValues) {
     this(dataType, Arrays.asList(rawdataValues));
   }
@@ -72,110 +81,213 @@ public class BeamRecord implements Serializable {
     dataValues.set(index, fieldValue);
   }
 
+  /**
+   * Get value by field name.
+   */
   public Object getFieldValue(String fieldName) {
     return getFieldValue(dataType.getFieldNames().indexOf(fieldName));
   }
 
+  /**
+   * Get a {@link Byte} value by field name, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Byte getByte(String fieldName) {
     return (Byte) getFieldValue(fieldName);
   }
 
+  /**
+   * Get a {@link Short} value by field name, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Short getShort(String fieldName) {
     return (Short) getFieldValue(fieldName);
   }
 
+  /**
+   * Get a {@link Integer} value by field name, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Integer getInteger(String fieldName) {
     return (Integer) getFieldValue(fieldName);
   }
 
+  /**
+   * Get a {@link Float} value by field name, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Float getFloat(String fieldName) {
     return (Float) getFieldValue(fieldName);
   }
 
+  /**
+   * Get a {@link Double} value by field name, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Double getDouble(String fieldName) {
     return (Double) getFieldValue(fieldName);
   }
 
+  /**
+   * Get a {@link Long} value by field name, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Long getLong(String fieldName) {
     return (Long) getFieldValue(fieldName);
   }
 
+  /**
+   * Get a {@link String} value by field name, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public String getString(String fieldName) {
     return (String) getFieldValue(fieldName);
   }
 
+  /**
+   * Get a {@link Date} value by field name, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Date getDate(String fieldName) {
     return (Date) getFieldValue(fieldName);
   }
 
+  /**
+   * Get a {@link GregorianCalendar} value by field name, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public GregorianCalendar getGregorianCalendar(String fieldName) {
     return (GregorianCalendar) getFieldValue(fieldName);
   }
 
+  /**
+   * Get a {@link BigDecimal} value by field name, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public BigDecimal getBigDecimal(String fieldName) {
     return (BigDecimal) getFieldValue(fieldName);
   }
 
+  /**
+   * Get a {@link Boolean} value by field name, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Boolean getBoolean(String fieldName) {
     return (Boolean) getFieldValue(fieldName);
   }
 
+  /**
+   * Get value by field index.
+   */
   public Object getFieldValue(int fieldIdx) {
     return dataValues.get(fieldIdx);
   }
 
+  /**
+   * Get a {@link Byte} value by field index, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Byte getByte(int idx) {
     return (Byte) getFieldValue(idx);
   }
 
+  /**
+   * Get a {@link Short} value by field index, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Short getShort(int idx) {
     return (Short) getFieldValue(idx);
   }
 
+  /**
+   * Get a {@link Integer} value by field index, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Integer getInteger(int idx) {
     return (Integer) getFieldValue(idx);
   }
 
+  /**
+   * Get a {@link Float} value by field index, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Float getFloat(int idx) {
     return (Float) getFieldValue(idx);
   }
 
+  /**
+   * Get a {@link Double} value by field index, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Double getDouble(int idx) {
     return (Double) getFieldValue(idx);
   }
 
+  /**
+   * Get a {@link Long} value by field index, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Long getLong(int idx) {
     return (Long) getFieldValue(idx);
   }
 
+  /**
+   * Get a {@link String} value by field index, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public String getString(int idx) {
     return (String) getFieldValue(idx);
   }
 
+  /**
+   * Get a {@link Date} value by field index, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Date getDate(int idx) {
     return (Date) getFieldValue(idx);
   }
 
+  /**
+   * Get a {@link GregorianCalendar} value by field index, {@link ClassCastException} is
thrown
+   * if type doesn't match.
+   */
   public GregorianCalendar getGregorianCalendar(int idx) {
     return (GregorianCalendar) getFieldValue(idx);
   }
 
+  /**
+   * Get a {@link BigDecimal} value by field index, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public BigDecimal getBigDecimal(int idx) {
     return (BigDecimal) getFieldValue(idx);
   }
 
+  /**
+   * Get a {@link Boolean} value by field index, {@link ClassCastException} is thrown
+   * if type doesn't match.
+   */
   public Boolean getBoolean(int idx) {
     return (Boolean) getFieldValue(idx);
   }
 
+  /**
+   * Return the size of data fields.
+   */
   public int getFieldCount() {
     return dataValues.size();
   }
 
+  /**
+   * Return the list of data values.
+   */
   public List<Object> getDataValues() {
     return dataValues;
   }
 
+  /**
+   * Return {@link BeamRecordType} which describes the fields.
+   */
   public BeamRecordType getDataType() {
     return dataType;
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/cd27036a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/BeamRecordType.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/BeamRecordType.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/BeamRecordType.java
index 29cc80d..620361c 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/BeamRecordType.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/BeamRecordType.java
@@ -25,14 +25,22 @@ import org.apache.beam.sdk.coders.BeamRecordCoder;
 import org.apache.beam.sdk.coders.Coder;
 
 /**
- * The default type provider used in {@link BeamRecord}.
+ * {@link BeamRecordType} describes the fields in {@link BeamRecord}, extra checking can
be added
+ * by overwriting {@link BeamRecordType#validateValueType(int, Object)}.
  */
 @Experimental
 public class BeamRecordType implements Serializable{
   private List<String> fieldNames;
   private List<Coder> fieldCoders;
 
+  /**
+   * Create a {@link BeamRecordType} with a name and Coder for each field.
+   */
   public BeamRecordType(List<String> fieldNames, List<Coder> fieldCoders) {
+    if (fieldNames.size() != fieldCoders.size()) {
+      throw new IllegalStateException(
+          "the size of fieldNames and fieldCoders need to be the same.");
+    }
     this.fieldNames = fieldNames;
     this.fieldCoders = fieldCoders;
   }
@@ -41,30 +49,42 @@ public class BeamRecordType implements Serializable{
    * Validate input fieldValue for a field.
    * @throws IllegalArgumentException throw exception when the validation fails.
    */
-   public void validateValueType(int index, Object fieldValue)
-      throws IllegalArgumentException{
-     //do nothing by default.
-   }
+  public void validateValueType(int index, Object fieldValue)
+     throws IllegalArgumentException{
+    //do nothing by default.
+  }
 
-   /**
-    * Get the coder for {@link BeamRecordCoder}.
-    */
-   public BeamRecordCoder getRecordCoder(){
-     return BeamRecordCoder.of(this, fieldCoders);
-   }
+  /**
+   * Return the coder for {@link BeamRecord}, which wraps {@link #fieldCoders} for each field.
+   */
+  public BeamRecordCoder getRecordCoder(){
+    return BeamRecordCoder.of(this, fieldCoders);
+  }
 
-   public List<String> getFieldNames(){
-     return ImmutableList.copyOf(fieldNames);
-   }
+  /**
+   * Returns an immutable list of field names.
+   */
+  public List<String> getFieldNames(){
+    return ImmutableList.copyOf(fieldNames);
+  }
 
-   public String getFieldNameByIndex(int index){
-     return fieldNames.get(index);
-   }
+  /**
+   * Return the name of field by index.
+   */
+  public String getFieldNameByIndex(int index){
+    return fieldNames.get(index);
+  }
 
-   public int findIndexOfField(String fieldName){
-     return fieldNames.indexOf(fieldName);
-   }
+  /**
+   * Find the index of a given field.
+   */
+  public int findIndexOfField(String fieldName){
+    return fieldNames.indexOf(fieldName);
+  }
 
+  /**
+   * Return the count of fields.
+   */
   public int getFieldCount(){
     return fieldNames.size();
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/cd27036a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/interpreter/operator/BeamSqlUdfExpression.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/interpreter/operator/BeamSqlUdfExpression.java
b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/interpreter/operator/BeamSqlUdfExpression.java
index 123e6a0..625de2c 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/interpreter/operator/BeamSqlUdfExpression.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/interpreter/operator/BeamSqlUdfExpression.java
@@ -18,6 +18,7 @@
 package org.apache.beam.sdk.extensions.sql.impl.interpreter.operator;
 
 import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
 import java.util.ArrayList;
 import java.util.List;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
@@ -79,8 +80,10 @@ public class BeamSqlUdfExpression extends BeamSqlExpression {
       for (String pc : paraClassName) {
         paraClass.add(Class.forName(pc));
       }
-      udfIns = Class.forName(className).newInstance();
       method = Class.forName(className).getMethod(methodName, paraClass.toArray(new Class<?>[]
{}));
+      if (!Modifier.isStatic(method.getModifiers())) {
+        udfIns = Class.forName(className).newInstance();
+      }
     } catch (Exception e) {
       throw new RuntimeException(e);
     }


Mime
View raw message