Added pass through view support

Author: davidrabinowitz

spark-bigquery-dsv2/spark-3.5-bigquery-lib/src/main/java/com/google/cloud/spark/bigquery/BigQueryCatalogExtension.java

@@ -22,8 +22,10 @@
 import org.apache.spark.sql.catalyst.analysis.NoSuchFunctionException;
 import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
 import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.analysis.NoSuchViewException;
 import org.apache.spark.sql.catalyst.analysis.NonEmptyNamespaceException;
 import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.ViewAlreadyExistsException;
 import org.apache.spark.sql.connector.catalog.CatalogExtension;
 import org.apache.spark.sql.connector.catalog.CatalogPlugin;
 import org.apache.spark.sql.connector.catalog.FunctionCatalog;
@@ -33,6 +35,9 @@
 import org.apache.spark.sql.connector.catalog.Table;
 import org.apache.spark.sql.connector.catalog.TableCatalog;
 import org.apache.spark.sql.connector.catalog.TableChange;
+import org.apache.spark.sql.connector.catalog.View;
+import org.apache.spark.sql.connector.catalog.ViewCatalog;
+import org.apache.spark.sql.connector.catalog.ViewChange;
 import org.apache.spark.sql.connector.catalog.functions.UnboundFunction;
 import org.apache.spark.sql.connector.expressions.Transform;
 import org.apache.spark.sql.internal.SQLConf;
@@ -41,7 +46,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class BigQueryCatalogExtension implements CatalogExtension {
+public class BigQueryCatalogExtension implements CatalogExtension, ViewCatalog {
 
   private static final Logger logger = LoggerFactory.getLogger(BigQueryCatalogExtension.class);
   private static final String[] DEFAULT_NAMESPACE = new String[] {"default"};
@@ -166,6 +171,68 @@ public boolean dropTable(Identifier ident) {
   public void renameTable(Identifier oldIdent, Identifier newIdent)
       throws NoSuchTableException, TableAlreadyExistsException {}
 
+  @Override
+  public Identifier[] listViews(String... namespace) throws NoSuchNamespaceException {
+    return delegateAsViewCatalog().listViews(namespace);
+  }
+
+  @Override
+  public View loadView(Identifier ident) throws NoSuchViewException {
+    return delegateAsViewCatalog().loadView(ident);
+  }
+
+  @Override
+  public void invalidateView(Identifier ident) {
+    delegateAsViewCatalog().invalidateView(ident);
+  }
+
+  @Override
+  public boolean viewExists(Identifier ident) {
+    return delegateAsViewCatalog().viewExists(ident);
+  }
+
+  @Override
+  public View createView(
+      Identifier ident,
+      String sql,
+      String currentCatalog,
+      String[] currentNamespace,
+      StructType schema,
+      String[] queryColumnNames,
+      String[] columnAliases,
+      String[] columnComments,
+      Map<String, String> properties)
+      throws ViewAlreadyExistsException, NoSuchNamespaceException {
+    return delegateAsViewCatalog()
+        .createView(
+            ident,
+            sql,
+            currentCatalog,
+            currentNamespace,
+            schema,
+            queryColumnNames,
+            columnAliases,
+            columnComments,
+            properties);
+  }
+
+  @Override
+  public View alterView(Identifier ident, ViewChange... changes)
+      throws NoSuchViewException, IllegalArgumentException {
+    return delegateAsViewCatalog().alterView(ident, changes);
+  }
+
+  @Override
+  public boolean dropView(Identifier ident) {
+    return delegateAsViewCatalog().dropView(ident);
+  }
+
+  @Override
+  public void renameView(Identifier oldIdent, Identifier newIdent)
+      throws NoSuchViewException, ViewAlreadyExistsException {
+    delegateAsViewCatalog().renameView(oldIdent, newIdent);
+  }
+
   private TableCatalog delegateAsTableCatalog() {
     return (TableCatalog) sessionCatalog;
   }
@@ -177,4 +244,8 @@ private FunctionCatalog delegateAsFunctionCatalog() {
   private SupportsNamespaces delegateAsSupportsNamespaces() {
     return (SupportsNamespaces) sessionCatalog;
   }
+
+  private ViewCatalog delegateAsViewCatalog() {
+    return (ViewCatalog) sessionCatalog;
+  }
 }

spark-bigquery-dsv2/spark-3.5-bigquery/src/test/java/com/google/cloud/spark/bigquery/integration/CatalogExtensionIntegrationTest.java

@@ -15,10 +15,37 @@
  */
 package com.google.cloud.spark.bigquery.integration;
 
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.bigquery.Table;
+import com.google.cloud.bigquery.TableId;
+import java.util.List;
+import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SparkSession;
+import org.junit.Test;
 
 public class CatalogExtensionIntegrationTest extends CatalogIntegrationTestBase {
 
+  @Test
+  public void testCreateGlobalTemporaryView() throws Exception {
+    String dataset = testDataset.testDataset;
+    String view = "test_view_" + System.nanoTime();
+
+    try (SparkSession spark = createSparkSession()) {
+      // prepare the table
+      spark.sql("CREATE TABLE " + fullTableName(dataset) + " AS SELECT 1 AS id, 'foo' AS data;");
+      Table table = bigquery.getTable(TableId.of(dataset, testTable));
+      assertThat(table).isNotNull();
+      assertThat(selectCountStarFrom(dataset, testTable)).isEqualTo(1L);
+
+      spark.sql(
+          String.format(
+              "CREATE GLOBAL TEMPORARY VIEW %s AS select * from %s", view, fullTableName(dataset)));
+      List<Row> result = spark.sql("SELECT * FROM global_temp." + view).collectAsList();
+      assertThat(result).hasSize(1);
+    }
+  }
+
   @Override
   protected SparkSession createSparkSession() {
     return SparkSession.builder()

spark-bigquery-dsv2/spark-3.5-bigquery/src/test/java/com/google/cloud/spark/bigquery/integration/CatalogIntegrationTestBase.java

@@ -40,7 +40,7 @@ public abstract class CatalogIntegrationTestBase {
 
   BigQuery bigquery = IntegrationTestUtils.getBigquery();
 
-  private String testTable;
+  protected String testTable;
 
   @Before
   public void renameTestTable() {
@@ -168,15 +168,15 @@ private void internalTestCreateTableWithExplicitTarget(String dataset)
     }
   }
 
-  private String fullTableName(String dataset) {
+  protected String fullTableName(String dataset) {
     return dataset.equals(DEFAULT_NAMESPACE)
         ? "`" + testTable + "`"
         : "`" + dataset + "`.`" + testTable + "`";
   }
 
   // this is needed as with direct write the table's metadata can e updated only after few minutes.
   // Queries take pending data into account though.
-  private long selectCountStarFrom(String dataset, String table) throws InterruptedException {
+  protected long selectCountStarFrom(String dataset, String table) throws InterruptedException {
     return bigquery
         .query(
             QueryJobConfiguration.of(