[bigquery] Always escape table names

clients/bigquery/bigquery_test.go

@@ -17,15 +17,15 @@ func TestTempTableName(t *testing.T) {
 	trimTTL := func(tableName string) string {
 		lastUnderscore := strings.LastIndex(tableName, "_")
 		assert.GreaterOrEqual(t, lastUnderscore, 0)
-		epoch, err := strconv.ParseInt(tableName[lastUnderscore+1:], 10, 64)
+		epoch, err := strconv.ParseInt(tableName[lastUnderscore+1:len(tableName)-1], 10, 64)
 		assert.NoError(t, err)
 		assert.Greater(t, time.Unix(epoch, 0), time.Now().Add(5*time.Hour)) // default TTL is 6 hours from now
-		return tableName[:lastUnderscore]
+		return tableName[:lastUnderscore] + string(tableName[len(tableName)-1])
 	}
 
 	store := &Store{config: config.Config{BigQuery: &config.BigQuery{ProjectID: "123454321"}}}
 	tableData := optimization.NewTableData(nil, config.Replication, nil, kafkalib.TopicConfig{Database: "db", Schema: "schema"}, "table")
 	tableID := store.IdentifierFor(tableData.TopicConfig(), tableData.Name())
 	tempTableName := shared.TempTableID(tableID, "sUfFiX").FullyQualifiedName()
-	assert.Equal(t, "`123454321`.`db`.table___artie_sUfFiX", trimTTL(tempTableName))
+	assert.Equal(t, "`123454321`.`db`.`table___artie_sUfFiX`", trimTTL(tempTableName))
 }

clients/bigquery/merge_test.go

@@ -41,20 +41,20 @@ func (b *BigQueryTestSuite) TestBackfillColumn() {
 		{
 			name:        "col that has default value that needs to be backfilled (boolean)",
 			col:         needsBackfillCol,
-			backfillSQL: "UPDATE `db`.`public`.tableName SET foo = true WHERE foo IS NULL;",
-			commentSQL:  "ALTER TABLE `db`.`public`.tableName ALTER COLUMN foo SET OPTIONS (description=`{\"backfilled\": true}`);",
+			backfillSQL: "UPDATE `db`.`public`.`tableName` SET foo = true WHERE foo IS NULL;",
+			commentSQL:  "ALTER TABLE `db`.`public`.`tableName` ALTER COLUMN foo SET OPTIONS (description=`{\"backfilled\": true}`);",
 		},
 		{
 			name:        "col that has default value that needs to be backfilled (string)",
 			col:         needsBackfillColStr,
-			backfillSQL: "UPDATE `db`.`public`.tableName SET foo2 = 'hello there' WHERE foo2 IS NULL;",
-			commentSQL:  "ALTER TABLE `db`.`public`.tableName ALTER COLUMN foo2 SET OPTIONS (description=`{\"backfilled\": true}`);",
+			backfillSQL: "UPDATE `db`.`public`.`tableName` SET foo2 = 'hello there' WHERE foo2 IS NULL;",
+			commentSQL:  "ALTER TABLE `db`.`public`.`tableName` ALTER COLUMN foo2 SET OPTIONS (description=`{\"backfilled\": true}`);",
 		},
 		{
 			name:        "col that has default value that needs to be backfilled (number)",
 			col:         needsBackfillColNum,
-			backfillSQL: "UPDATE `db`.`public`.tableName SET foo3 = 3.5 WHERE foo3 IS NULL;",
-			commentSQL:  "ALTER TABLE `db`.`public`.tableName ALTER COLUMN foo3 SET OPTIONS (description=`{\"backfilled\": true}`);",
+			backfillSQL: "UPDATE `db`.`public`.`tableName` SET foo3 = 3.5 WHERE foo3 IS NULL;",
+			commentSQL:  "ALTER TABLE `db`.`public`.`tableName` ALTER COLUMN foo3 SET OPTIONS (description=`{\"backfilled\": true}`);",
 		},
 	}
 

clients/bigquery/tableid.go

@@ -3,9 +3,7 @@ package bigquery
 import (
 	"fmt"
 
-	"github.com/artie-labs/transfer/lib/config/constants"
 	"github.com/artie-labs/transfer/lib/destination/types"
-	"github.com/artie-labs/transfer/lib/sql"
 )
 
 type TableIdentifier struct {
@@ -40,11 +38,6 @@ func (ti TableIdentifier) WithTable(table string) types.TableIdentifier {
 
 func (ti TableIdentifier) FullyQualifiedName() string {
 	// The fully qualified name for BigQuery is: project_id.dataset.tableName.
-	// We are escaping the project_id and dataset because there could be special characters.
-	return fmt.Sprintf(
-		"`%s`.`%s`.%s",
-		ti.projectID,
-		ti.dataset,
-		sql.EscapeNameIfNecessary(ti.table, false, constants.BigQuery),
-	)
+	// We are escaping the project_id, dataset, and table because there could be special characters.
+	return fmt.Sprintf("`%s`.`%s`.`%s`", ti.projectID, ti.dataset, ti.table)
 }

clients/bigquery/tableid_test.go

@@ -17,9 +17,9 @@ func TestTableIdentifier_WithTable(t *testing.T) {
 }
 
 func TestTableIdentifier_FullyQualifiedName(t *testing.T) {
-	// Table name that does not need escaping:
-	assert.Equal(t, "`project`.`dataset`.foo", NewTableIdentifier("project", "dataset", "foo").FullyQualifiedName())
+	// Table name that is not a reserved word:
+	assert.Equal(t, "`project`.`dataset`.`foo`", NewTableIdentifier("project", "dataset", "foo").FullyQualifiedName())
 
-	// Table name that needs escaping:
+	// Table name that is a reserved word:
 	assert.Equal(t, "`project`.`dataset`.`table`", NewTableIdentifier("project", "dataset", "table").FullyQualifiedName())
 }

lib/destination/ddl/ddl_temp_test.go

@@ -115,6 +115,6 @@ func (d *DDLTestSuite) TestCreateTemporaryTable() {
 		assert.Equal(d.T(), 1, d.fakeBigQueryStore.ExecCallCount())
 		bqQuery, _ := d.fakeBigQueryStore.ExecArgsForCall(0)
 		// Cutting off the expiration_timestamp since it's time based.
-		assert.Contains(d.T(), bqQuery, "CREATE TABLE IF NOT EXISTS `db`.`schema`.tempTableName (foo string,bar float64,`select` string) OPTIONS (expiration_timestamp =")
+		assert.Contains(d.T(), bqQuery, "CREATE TABLE IF NOT EXISTS `db`.`schema`.`tempTableName` (foo string,bar float64,`select` string) OPTIONS (expiration_timestamp =")
 	}
 }