Update.

clients/databricks/dialect/dialect.go

@@ -4,12 +4,11 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/artie-labs/transfer/lib/typing/ext"
-
 	"github.com/artie-labs/transfer/lib/config/constants"
 	"github.com/artie-labs/transfer/lib/sql"
 	"github.com/artie-labs/transfer/lib/typing"
 	"github.com/artie-labs/transfer/lib/typing/columns"
+	"github.com/artie-labs/transfer/lib/typing/ext"
 )
 
 type DatabricksDialect struct{}

clients/databricks/store.go

@@ -1,6 +1,7 @@
 package databricks
 
 import (
+	"github.com/artie-labs/transfer/clients/databricks/dialect"
 	"github.com/artie-labs/transfer/clients/shared"
 	"github.com/artie-labs/transfer/lib/config"
 	"github.com/artie-labs/transfer/lib/db"
@@ -24,11 +25,11 @@ func (s Store) Append(tableData *optimization.TableData, useTempTable bool) erro
 }
 
 func (s Store) IdentifierFor(topicConfig kafkalib.TopicConfig, table string) sql.TableIdentifier {
-	//return NewTableIdentifier(topicConfig.Schema, table)
+	return NewTableIdentifier(topicConfig.Database, topicConfig.Schema, table)
 }
 
 func (s Store) Dialect() sql.Dialect {
-	//return dialect.DatabricksDialect{}
+	return dialect.DatabricksDialect{}
 }
 
 func (s Store) Dedupe(tableID sql.TableIdentifier, primaryKeys []string, includeArtieUpdatedAt bool) error {

clients/databricks/tableid.go

@@ -0,0 +1,48 @@
+package databricks
+
+import (
+	"fmt"
+
+	"github.com/artie-labs/transfer/clients/databricks/dialect"
+	"github.com/artie-labs/transfer/lib/sql"
+)
+
+var _dialect = dialect.DatabricksDialect{}
+
+type TableIdentifier struct {
+	database string
+	schema   string
+	table    string
+}
+
+func NewTableIdentifier(database, schema, table string) TableIdentifier {
+	return TableIdentifier{
+		database: database,
+		schema:   schema,
+		table:    table,
+	}
+}
+
+func (ti TableIdentifier) Database() string {
+	return ti.database
+}
+
+func (ti TableIdentifier) Schema() string {
+	return ti.schema
+}
+
+func (ti TableIdentifier) EscapedTable() string {
+	return _dialect.QuoteIdentifier(ti.table)
+}
+
+func (ti TableIdentifier) Table() string {
+	return ti.table
+}
+
+func (ti TableIdentifier) WithTable(table string) sql.TableIdentifier {
+	return NewTableIdentifier(ti.database, ti.schema, table)
+}
+
+func (ti TableIdentifier) FullyQualifiedName() string {
+	return fmt.Sprintf("%s.%s.%s", ti.database, ti.schema, ti.EscapedTable())
+}