Add MATERIALIZED_VIEW as table type

connector/src/main/java/com/google/cloud/bigquery/connector/common/BigQueryClient.java

@@ -73,7 +73,7 @@ public TableInfo getSupportedTable(TableId tableId, boolean viewsEnabled, String
         if (TableDefinition.Type.TABLE == tableType) {
             return table;
         }
-        if (TableDefinition.Type.VIEW == tableType) {
+        if (TableDefinition.Type.VIEW == tableType || TableDefinition.Type.MATERIALIZED_VIEW == tableType) {
             if (viewsEnabled) {
                 return table;
             } else {
@@ -167,7 +167,7 @@ public long calculateTableSize(TableInfo tableInfo, Optional<String> filter) {
             TableDefinition.Type type = tableInfo.getDefinition().getType();
             if (type == TableDefinition.Type.TABLE && !filter.isPresent()) {
                 return tableInfo.getNumRows().longValue();
-            } else if (type == TableDefinition.Type.VIEW ||
+            } else if (type == TableDefinition.Type.VIEW || type == TableDefinition.Type.MATERIALIZED_VIEW ||
                     (type == TableDefinition.Type.TABLE && filter.isPresent())) {
                 // run a query
                 String table = fullTableName(tableInfo.getTableId());

connector/src/main/java/com/google/cloud/bigquery/connector/common/ReadSessionCreator.java

@@ -126,7 +126,7 @@ TableInfo getActualTable(
         if (TableDefinition.Type.TABLE == tableType) {
             return table;
         }
-        if (TableDefinition.Type.VIEW == tableType) {
+        if (TableDefinition.Type.VIEW == tableType || TableDefinition.Type.MATERIALIZED_VIEW == tableType) {
             if (!config.viewsEnabled) {
                 throw new BigQueryConnectorException(UNSUPPORTED, format(
                         "Views are not enabled. You can enable views by setting '%s' to true. Notice additional cost may occur.",

connector/src/main/scala/com/google/cloud/spark/bigquery/BigQueryRelationProvider.scala

@@ -16,7 +16,7 @@
 package com.google.cloud.spark.bigquery
 
 import com.google.auth.Credentials
-import com.google.cloud.bigquery.TableDefinition.Type.{TABLE, VIEW}
+import com.google.cloud.bigquery.TableDefinition.Type.{MATERIALIZED_VIEW, TABLE, VIEW}
 import com.google.cloud.bigquery.{BigQuery, BigQueryOptions, TableDefinition}
 import com.google.cloud.spark.bigquery.direct.DirectBigQueryRelation
 import org.apache.spark.sql.sources._
@@ -57,7 +57,7 @@ class BigQueryRelationProvider(
       .getOrElse(sys.error(s"Table $tableName not found"))
     table.getDefinition[TableDefinition].getType match {
       case TABLE => new DirectBigQueryRelation(opts, table)(sqlContext)
-      case VIEW => if (opts.viewsEnabled) {
+      case VIEW | MATERIALIZED_VIEW => if (opts.viewsEnabled) {
         new DirectBigQueryRelation(opts, table)(sqlContext)
       } else {
         sys.error(

connector/src/test/scala/com/google/cloud/spark/bigquery/it/SparkBigQueryEndToEndITSuite.scala

@@ -346,7 +346,7 @@ class SparkBigQueryEndToEndITSuite extends FunSuite
       countResults should equal(countAfterCollect)
     }
     */
-    
+
     test("read data types. DataSource %s".format(dataSourceFormat)) {
       val allTypesTable = readAllTypesTable(dataSourceFormat)
       val expectedRow = spark.range(1).select(TestConstants.ALL_TYPES_TABLE_COLS: _*).head.toSeq
@@ -542,6 +542,15 @@ class SparkBigQueryEndToEndITSuite extends FunSuite
     assert(df.schema == allTypesTable.schema)
   }
 
+  test("query materialized view") {
+    var df = spark.read.format("bigquery")
+      .option("table", "bigquery-public-data:ethereum_blockchain.live_logs")
+      .option("viewsEnabled", "true")
+      .option("viewMaterializationProject", "bigquery-public-data")
+      .option("viewMaterializationDataset", "ethereum_blockchain")
+      .load()
+  }
+
   test("write to bq - adding the settings to spark.conf" ) {
     spark.conf.set("temporaryGcsBucket", temporaryGcsBucket)
     val df = initialData