Replace the deprecated method to read BigQuery table (#10)

* Replace the deprecated method to read BigQuery

* Change the version to 0.2.1

* Modify README

* Fix

README.md

@@ -54,7 +54,7 @@ It is horizontally-scalable on top of distributed system, since apache beam can
 mvn clean package
 
 # Run bigquery-to-datastore via the compiled JAR file
-java -cp $(pwd)/target/kuromoji-for-bigquery-bundled-0.2.0.jar \
+java -cp $(pwd)/target/kuromoji-for-bigquery-bundled-0.2.1.jar \
   com.github.yuiskw.beam.Kuromoji4BigQuery \
   --project=test-project-id
   --schema=id:integer
@@ -71,6 +71,12 @@ java -cp $(pwd)/target/kuromoji-for-bigquery-bundled-0.2.0.jar \
   --workerMachineType=n1-standard-2
 ```
 
+## Versions
+|kuromoji-for-bigquery|Apache Beam|kuromoji|
+|---------------------|-----------|--------|
+|0.1.0                |2.1.0      |0.7.7   |
+|0.2.x                |2.20.0     |0.7.7   |
+
 ## License
 
 Copyright (c) 2017 Yu Ishikawa.

pom.xml

@@ -22,7 +22,7 @@
 
   <groupId>com.github.yuiskw</groupId>
   <artifactId>kuromoji-for-bigquery</artifactId>
-  <version>0.2.0</version>
+  <version>0.2.1</version>
 
   <packaging>jar</packaging>
 

src/main/java/com/github/yuiskw/beam/Kuromoji4BigQuery.java

@@ -110,7 +110,10 @@ public static void main(String[] args) {
         .setProjectId(projectId)
         .setDatasetId(inputDatasetId)
         .setTableId(inputTableId);
-    BigQueryIO.Read reader = BigQueryIO.read().from(inputTableRef);
+    BigQueryIO.TypedRead<TableRow> reader = BigQueryIO.readTableRows()
+        .withMethod(BigQueryIO.TypedRead.Method.DIRECT_READ)  // to use the BigQuery Storage API
+        .withSelectedFields(getSelectedFields(schemaMap, tokenizedColumn))
+        .from(inputTableRef);
 
     // Output
     TableReference outputTableRef = new TableReference()
@@ -214,4 +217,29 @@ public static TableSchema convertToTableSchema(
     TableSchema schema = new TableSchema().setFields(fields);
     return schema;
   }
+
+  /**
+   * Get selected fields
+   *
+   * @param schemaMap a map of BigQuery table schema
+   * @param tokenizedColumn column to be tokenized
+   * @return A list of selected fields
+   */
+  public static List<String> getSelectedFields(
+      LinkedHashMap<String, String> schemaMap,
+      String tokenizedColumn) {
+    ArrayList<String> columns = new ArrayList<String>();
+
+    // Extract columns names from the schema.
+    String[] columnsInSchema = schemaMap.keySet().toArray(new String[schemaMap.keySet().size()]);
+    for (String c: columnsInSchema) {
+      columns.add(c);
+    }
+
+    // Append the tokenized column, if it doesn't exist in the schema
+    if (! columns.contains(tokenizedColumn)) {
+      columns.add(tokenizedColumn);
+    }
+    return columns;
+  }
 }

src/test/java/com/github/yuiskw/beam/Kuromoji4BigQueryTest.java

@@ -3,6 +3,7 @@
  */
 package com.github.yuiskw.beam;
 
+import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.List;
 
@@ -80,6 +81,28 @@ public void testParseSchemaWithSpaces() {
     assertEquals("float", parents.get("height"));
   }
 
+  @Test
+  public void testGetSelectedFields1() {
+    String tokenizedColumn = "text";
+    LinkedHashMap<String, String> schemaMap = new LinkedHashMap<String, String>();
+    schemaMap.put("x", "interger");
+    schemaMap.put("y", "string");
+    List<String> selectedFields = Kuromoji4BigQuery.getSelectedFields(schemaMap, tokenizedColumn);
+    List<String> expected = Arrays.asList("x", "y", "text");
+    assertEquals(selectedFields, expected);
+  }
+
+  @Test
+  public void testGetSelectedFields2() {
+    String tokenizedColumn = "text";
+    LinkedHashMap<String, String> schemaMap = new LinkedHashMap<String, String>();
+    schemaMap.put("x", "interger");
+    schemaMap.put("text", "string");
+    List<String> selectedFields = Kuromoji4BigQuery.getSelectedFields(schemaMap, tokenizedColumn);
+    List<String> expected = Arrays.asList("x", "text");
+    assertEquals(selectedFields, expected);
+  }
+
 
   /**
    Test Query