From de27f69270e1806fe3d49a15cbff2f229c858442 Mon Sep 17 00:00:00 2001
From: Robin Tang <rtang.cs@gmail.com>
Date: Sat, 14 Dec 2024 19:50:15 -0800
Subject: [PATCH] WIP.

---
 clients/databricks/dialect/dialect.go      |  9 +++++---
 clients/databricks/dialect/dialect_test.go | 24 ++++++++++++++++++++++
 2 files changed, 30 insertions(+), 3 deletions(-)

diff --git a/clients/databricks/dialect/dialect.go b/clients/databricks/dialect/dialect.go
index 1a54254f8..b4b8fdb5b 100644
--- a/clients/databricks/dialect/dialect.go
+++ b/clients/databricks/dialect/dialect.go
@@ -31,11 +31,14 @@ func (DatabricksDialect) IsTableDoesNotExistErr(err error) bool {
 }
 
 func (d DatabricksDialect) BuildIsNotToastValueExpression(tableAlias constants.TableAlias, column columns.Column) string {
+	toastedValue := "%" + constants.ToastUnavailableValuePlaceholder + "%"
 	colName := sql.QuoteTableAliasColumn(tableAlias, column, d)
-	if column.KindDetails == typing.Struct {
-		return fmt.Sprintf("COALESCE(%s != {'key': '%s'}, true)", colName, constants.ToastUnavailableValuePlaceholder)
+	switch column.KindDetails {
+	case typing.String:
+		return fmt.Sprintf("COALESCE(%s NOT LIKE '%s', TRUE)", colName, toastedValue)
+	default:
+		return fmt.Sprintf("COALESCE(TO_VARCHAR(%s) NOT LIKE '%s', TRUE)", colName, toastedValue)
 	}
-	return fmt.Sprintf("COALESCE(%s != '%s', true)", colName, constants.ToastUnavailableValuePlaceholder)
 }
 
 func (DatabricksDialect) BuildDedupeTableQuery(tableID sql.TableIdentifier, primaryKeys []string) string {
diff --git a/clients/databricks/dialect/dialect_test.go b/clients/databricks/dialect/dialect_test.go
index 6d3c3fc59..053b7a7e4 100644
--- a/clients/databricks/dialect/dialect_test.go
+++ b/clients/databricks/dialect/dialect_test.go
@@ -50,6 +50,30 @@ func TestDatabricksDialect_IsTableDoesNotExistErr(t *testing.T) {
 	}
 }
 
+func TestDatabricks_BuildIsNotToastValueExpression(t *testing.T) {
+	{
+		// Unspecified data type
+		assert.Equal(t,
+			"COALESCE(TO_VARCHAR(tbl.`bar`) NOT LIKE '%__debezium_unavailable_value%', TRUE)",
+			DatabricksDialect{}.BuildIsNotToastValueExpression("tbl", columns.NewColumn("bar", typing.Invalid)),
+		)
+	}
+	{
+		// Structs
+		assert.Equal(t,
+			"COALESCE(TO_VARCHAR(tbl.`foo`) NOT LIKE '%__debezium_unavailable_value%', TRUE)",
+			DatabricksDialect{}.BuildIsNotToastValueExpression("tbl", columns.NewColumn("foo", typing.Struct)),
+		)
+	}
+	{
+		// String
+		assert.Equal(t,
+			"COALESCE(tbl.`bar` NOT LIKE '%__debezium_unavailable_value%', TRUE)",
+			DatabricksDialect{}.BuildIsNotToastValueExpression("tbl", columns.NewColumn("bar", typing.String)),
+		)
+	}
+}
+
 func TestDatabricksDialect_BuildCreateTableQuery(t *testing.T) {
 	fakeTableID := &mocks.FakeTableIdentifier{}
 	fakeTableID.FullyQualifiedNameReturns("{TABLE}")