Reduce code breaking change

Author: Michel Davit

Diff for: integration/src/test/scala/com/spotify/scio/bigquery/BigQueryClientIT.scala

@@ -154,7 +154,7 @@ class BigQueryClientIT extends AnyFlatSpec with Matchers {
 
   "TableService.getRows" should "work" in {
     val rows =
-      bq.tables.rows(Table("bigquery-public-data:samples.shakespeare")).take(10).toList
+      bq.tables.rows(Table.Spec("bigquery-public-data:samples.shakespeare")).take(10).toList
     val columns = Set("word", "word_count", "corpus", "corpus_date")
     all(rows.map(_.keySet().asScala)) shouldBe columns
   }

Diff for: integration/src/test/scala/com/spotify/scio/bigquery/TypedBigQueryIT.scala

@@ -98,7 +98,7 @@ object TypedBigQueryIT {
     val now = Instant.now().toString(TIME_FORMATTER)
     val spec =
       s"data-integration-test:bigquery_avro_it.$name${now}_${Random.nextInt(Int.MaxValue)}"
-    Table(spec)
+    Table.Spec(spec)
   }
   private val typedTable = table("records")
   private val tableRowTable = table("records_tablerow")

Diff for: integration/src/test/scala/com/spotify/scio/bigquery/types/BigQueryTypeIT.scala

@@ -187,9 +187,10 @@ class BigQueryTypeIT extends AnyFlatSpec with Matchers {
     tableReference.setProjectId("data-integration-test")
     tableReference.setDatasetId("partition_a")
     tableReference.setTableId("table_$LATEST")
-    Table(tableReference).latest().ref.getTableId shouldBe "table_20170302"
+    Table.Ref(tableReference).latest().ref.getTableId shouldBe "table_20170302"
 
-    Table("data-integration-test:partition_a.table_$LATEST")
+    Table
+      .Spec("data-integration-test:partition_a.table_$LATEST")
       .latest()
       .ref
       .getTableId shouldBe "table_20170302"

Diff for: scio-google-cloud-platform/src/main/scala/com/spotify/scio/bigquery/BigQueryIO.scala

@@ -143,13 +143,13 @@ object BigQueryIO {
       val selectedFields = bqt.selectedFields
       val rowRestriction = bqt.rowRestriction
       if (selectedFields.isEmpty && rowRestriction.isEmpty) {
-        Table(bqt.table.get)
+        Table.Spec(bqt.table.get)
       } else {
         val filter = Table.Filter(selectedFields.getOrElse(Nil), rowRestriction)
-        Table(bqt.table.get, filter)
+        Table.Spec(bqt.table.get, filter)
       }
     } else {
-      Table(bqt.table.get)
+      Table.Spec(bqt.table.get)
     }
     BigQueryIO(source)
   }

Diff for: scio-google-cloud-platform/src/main/scala/com/spotify/scio/bigquery/BigQueryTypes.scala

@@ -32,17 +32,16 @@ import org.joda.time.format.{DateTimeFormat, DateTimeFormatterBuilder}
 
 import java.math.MathContext
 import java.nio.ByteBuffer
+import scala.annotation.unused
 import scala.jdk.CollectionConverters._
 
 sealed trait Source {
-  protected type Impl <: Source
-  def latest(bq: BigQuery): Impl
-  def latest(): Impl = latest(BigQuery.defaultInstance())
+  def latest(bq: BigQuery): Source
+  def latest(): Source = latest(BigQuery.defaultInstance())
 }
 
 /** A wrapper type [[Query]] which wraps a SQL String. */
 final case class Query(underlying: String) extends Source {
-  override protected type Impl = Query
 
   /**
    * A helper method to replace the "$LATEST" placeholder in query to the latest common partition.
@@ -73,21 +72,22 @@ final case class Query(underlying: String) extends Source {
 }
 
 /**
- * Bigquery [[Table]]. Tables can be referenced by a table spec `String` or by a table reference
- * [[GTableReference]]. An additional [[Table.Filter]] can be given to specify selected fields and
- * row restrictions when used with the BQ storage read API.
+ * Bigquery [[Table]] abstracts the multiple ways of referencing Bigquery tables. Tables can be
+ * referenced by a table spec `String` or by a table reference [[GTableReference]]. An additional
+ * [[Table.Filter]] can be given to specify selected fields and row restrictions when used with the
+ * BQ storage read API.
  *
  * Example: Create a [[Table]] from a [[GTableReference]]:
  * {{{
  *   val tableReference = new TableReference
  *   tableReference.setProjectId("bigquery-public-data")
  *   tableReference.setDatasetId("samples")
  *   tableReference.setTableId("shakespeare")
- *   val table = Table(tableReference)
+ *   val table = Table.Ref(tableReference)
  * }}}
  * or with a spec string with filtering:
  * {{{
- *   val table = Table(
+ *   val table = Table.Spec(
  *     "bigquery-public-data:samples.shakespeare",
  *     List("word", "word_count"),
  *     "word_count > 10"
@@ -101,9 +101,8 @@ final case class Query(underlying: String) extends Source {
  * }}}
  */
 case class Table private (ref: GTableReference, filter: Option[Table.Filter]) extends Source {
-  override protected type Impl = Table
   lazy val spec: String = BigQueryHelpers.toTableSpec(ref)
-  def latest(bq: BigQuery): Table = {
+  override def latest(bq: BigQuery): Source = {
     val latestSpec = BigQueryPartitionUtil.latestTable(bq, spec)
     val latestRef = BigQueryHelpers.parseTableSpec(latestSpec)
     copy(latestRef)
@@ -151,40 +150,43 @@ object Table {
     rowRestriction: Option[String]
   )
 
-  def apply(ref: GTableReference): Table =
+  @unused private def apply(ref: GTableReference, filter: Option[Table.Filter]): Table =
+    new Table(ref, filter)
+
+  def Ref(ref: GTableReference): Table =
     new Table(ref, None)
 
-  def apply(ref: GTableReference, selectedFields: List[String]): Table =
+  def Ref(ref: GTableReference, selectedFields: List[String]): Table =
     new Table(ref, Some(Filter(selectedFields, None)))
 
-  def apply(ref: GTableReference, rowRestriction: String): Table =
+  def Ref(ref: GTableReference, rowRestriction: String): Table =
     new Table(ref, Some(Filter(List.empty, Some(rowRestriction))))
 
-  def apply(ref: GTableReference, selectedFields: List[String], rowRestriction: String): Table =
+  def Ref(ref: GTableReference, selectedFields: List[String], rowRestriction: String): Table =
     new Table(ref, Some(Filter(selectedFields, Some(rowRestriction))))
 
-  def apply(ref: GTableReference, filter: Table.Filter): Table =
+  def Ref(ref: GTableReference, filter: Table.Filter): Table =
     new Table(ref, Some(filter))
 
-  def apply(spec: String): Table =
+  def Spec(spec: String): Table =
     new Table(BigQueryHelpers.parseTableSpec(spec), None)
 
-  def apply(spec: String, selectedFields: List[String]): Table =
+  def Spec(spec: String, selectedFields: List[String]): Table =
     new Table(BigQueryHelpers.parseTableSpec(spec), Some(Filter(selectedFields, None)))
 
-  def apply(spec: String, rowRestriction: String): Table =
+  def Spec(spec: String, rowRestriction: String): Table =
     new Table(BigQueryHelpers.parseTableSpec(spec), Some(Filter(List.empty, Some(rowRestriction))))
 
-  def apply(spec: String, selectedFields: List[String], rowRestriction: String): Table =
+  def Spec(spec: String, selectedFields: List[String], rowRestriction: String): Table =
     new Table(
       BigQueryHelpers.parseTableSpec(spec),
       Some(Filter(selectedFields, Some(rowRestriction)))
     )
 
-  def apply(spec: String, filter: Table.Filter): Table =
+  def Spec(spec: String, filter: Table.Filter): Table =
     new Table(BigQueryHelpers.parseTableSpec(spec), Some(filter))
 
-  def apply(spec: String, filter: Option[Table.Filter]): Table =
+  def Spec(spec: String, filter: Option[Table.Filter]): Table =
     new Table(BigQueryHelpers.parseTableSpec(spec), filter)
 }
 

Diff for: scio-google-cloud-platform/src/main/scala/com/spotify/scio/bigquery/MockBigQuery.scala

@@ -181,7 +181,7 @@ class MockTable(
    */
   def withSample(numRows: Int): Unit = {
     ensureUnique()
-    val rows = bq.tables.rows(Table(original)).take(numRows).toList
+    val rows = bq.tables.rows(Table.Ref(original)).take(numRows).toList
     require(rows.length == numRows, s"Sample size ${rows.length} != requested $numRows")
     writeRows(rows)
     ()
@@ -193,7 +193,7 @@ class MockTable(
    */
   def withSample(minNumRows: Int, maxNumRows: Int): Unit = {
     ensureUnique()
-    val rows = bq.tables.rows(Table(original)).take(maxNumRows).toList
+    val rows = bq.tables.rows(Table.Ref(original)).take(maxNumRows).toList
     require(
       rows.length >= minNumRows && rows.length <= maxNumRows,
       s"Sample size ${rows.length} < requested minimal $minNumRows"

Diff for: scio-google-cloud-platform/src/main/scala/com/spotify/scio/bigquery/client/BigQuery.scala

@@ -89,7 +89,7 @@ final class BigQuery private (val client: Client) {
     val rows = if (newSource == null) {
       // newSource is missing, T's companion object must have either table or query
       if (bqt.isTable) {
-        tables.rows(STable(bqt.table.get))
+        tables.rows(STable.Spec(bqt.table.get))
       } else if (bqt.isQuery) {
         query.rows(bqt.queryRaw.get)
       } else {
@@ -98,7 +98,7 @@ final class BigQuery private (val client: Client) {
     } else {
       // newSource can be either table or query
       Try(BigQueryHelpers.parseTableSpec(newSource)).toOption
-        .map(STable.apply)
+        .map(STable.Ref)
         .map(tables.rows)
         .getOrElse(query.rows(newSource))
     }

Diff for: scio-google-cloud-platform/src/main/scala/com/spotify/scio/bigquery/client/QueryOps.scala

@@ -111,7 +111,7 @@ final private[client] class QueryOps(client: Client, tableService: TableOps, job
 
     newQueryJob(config).map { job =>
       jobService.waitForJobs(job)
-      tableService.rows(STable(job.table))
+      tableService.rows(STable.Ref(job.table))
     }.get
   }
 

Diff for: scio-google-cloud-platform/src/main/scala/com/spotify/scio/bigquery/taps.scala

@@ -79,7 +79,7 @@ final case class BigQueryTaps(self: Taps) {
       s"BigQuery Table: $table",
       () => bqc.tables.exists(table),
       () =>
-        BigQueryIO[TableRow](Table(table))
+        BigQueryIO[TableRow](Table.Ref(table))
           .tap(BigQueryIO.TableReadParam(BigQueryIO.Format.Default(), Method.DEFAULT))
     )
 
@@ -131,7 +131,7 @@ final case class BigQueryTaps(self: Taps) {
         val selectedFields = readOptions.getSelectedFieldsList.asScala.toList
         val rowRestriction = Option(readOptions.getRowRestriction)
         val filter = Table.Filter(selectedFields, rowRestriction)
-        val source = Table(table, filter)
+        val source = Table.Ref(table, filter)
         BigQueryIO[TableRow](source).tap(BigQueryIO.TableReadParam(format, Method.DIRECT_READ))
       }
     )
@@ -148,7 +148,7 @@ final case class BigQueryTaps(self: Taps) {
         val selectedFields = readOptions.getSelectedFieldsList.asScala.toList
         val rowRestriction = Option(readOptions.getRowRestriction)
         val filter = Table.Filter(selectedFields, rowRestriction)
-        val source = Table(table, filter)
+        val source = Table.Ref(table, filter)
         BigQueryIO[T](source).tap(BigQueryIO.TableReadParam(format, Method.DIRECT_READ))
       }
     )