From 4e0497b676ca94a58ac8c8a3c436558213c663e2 Mon Sep 17 00:00:00 2001 From: Robin Tang Date: Sun, 15 Dec 2024 21:18:54 -0800 Subject: [PATCH 1/4] Checkpoint. --- clients/databricks/store.go | 25 ++++++++++++++++++++++++- 1 file changed, 24 insertions(+), 1 deletion(-) diff --git a/clients/databricks/store.go b/clients/databricks/store.go index 02df68b94..1056e8dcc 100644 --- a/clients/databricks/store.go +++ b/clients/databricks/store.go @@ -7,6 +7,7 @@ import ( "log/slog" "os" "path/filepath" + "strings" _ "github.com/databricks/databricks-sql-go" "github.com/databricks/databricks-sql-go/driverctx" @@ -120,9 +121,31 @@ func (s Store) PrepareTemporaryTable(ctx context.Context, tableData *optimizatio } }() + var ordinalColumns []string + for idx, column := range tableData.ReadOnlyInMemoryCols().ValidColumns() { + ordinalColumn := fmt.Sprintf("_c%d", idx) + switch column.KindDetails.Kind { + case typing.Array.Kind: + ordinalColumn = fmt.Sprintf(`parse_json(%s)`, ordinalColumn) + } + + ordinalColumns = append(ordinalColumns, ordinalColumn) + } + // Copy file from DBFS -> table via COPY INTO, ref: https://docs.databricks.com/en/sql/language-manual/delta-copy-into.html // We'll need \\\\N here because we need to string escape. - copyCommand := fmt.Sprintf(`COPY INTO %s BY POSITION FROM '%s' FILEFORMAT = CSV FORMAT_OPTIONS ('escape' = '"', 'delimiter' = '\t', 'header' = 'false', 'nullValue' = '\\\\N')`, tempTableID.FullyQualifiedName(), file.DBFSFilePath()) + copyCommand := fmt.Sprintf(` +COPY INTO %s BY POSITION FROM (SELECT %s FROM '%s') FILEFORMAT = CSV FORMAT_OPTIONS ('escape' = '"', 'delimiter' = '\t', 'header' = 'false', 'nullValue' = '\\\\N')`, + // COPY INTO + tempTableID.FullyQualifiedName(), + // SELECT columns + strings.Join(ordinalColumns, ", "), + // FROM + file.DBFSFilePath(), + ) + + fmt.Println("copyCommand", copyCommand) + if _, err = s.ExecContext(ctx, copyCommand); err != nil { return fmt.Errorf("failed to run COPY INTO for temporary table: %w", err) } From 70c80e91d27f3ade9506f33686ec955da0075aac Mon Sep 17 00:00:00 2001 From: Robin Tang Date: Sun, 15 Dec 2024 21:52:04 -0800 Subject: [PATCH 2/4] Clean up. --- clients/databricks/store.go | 17 +++++++++++++---- lib/typing/values/string.go | 8 ++++++++ 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/clients/databricks/store.go b/clients/databricks/store.go index 1056e8dcc..563912479 100644 --- a/clients/databricks/store.go +++ b/clients/databricks/store.go @@ -126,7 +126,7 @@ func (s Store) PrepareTemporaryTable(ctx context.Context, tableData *optimizatio ordinalColumn := fmt.Sprintf("_c%d", idx) switch column.KindDetails.Kind { case typing.Array.Kind: - ordinalColumn = fmt.Sprintf(`parse_json(%s)`, ordinalColumn) + ordinalColumn = fmt.Sprintf(`PARSE_JSON(%s)`, ordinalColumn) } ordinalColumns = append(ordinalColumns, ordinalColumn) @@ -135,7 +135,18 @@ func (s Store) PrepareTemporaryTable(ctx context.Context, tableData *optimizatio // Copy file from DBFS -> table via COPY INTO, ref: https://docs.databricks.com/en/sql/language-manual/delta-copy-into.html // We'll need \\\\N here because we need to string escape. copyCommand := fmt.Sprintf(` -COPY INTO %s BY POSITION FROM (SELECT %s FROM '%s') FILEFORMAT = CSV FORMAT_OPTIONS ('escape' = '"', 'delimiter' = '\t', 'header' = 'false', 'nullValue' = '\\\\N')`, +COPY INTO %s +BY POSITION +FROM ( + SELECT %s FROM '%s' +) +FILEFORMAT = CSV +FORMAT_OPTIONS ( + 'escape' = '"', + 'delimiter' = '\t', + 'header' = 'false', + 'nullValue' = '\\\\N' +);`, // COPY INTO tempTableID.FullyQualifiedName(), // SELECT columns @@ -144,8 +155,6 @@ COPY INTO %s BY POSITION FROM (SELECT %s FROM '%s') FILEFORMAT = CSV FORMAT_OPTI file.DBFSFilePath(), ) - fmt.Println("copyCommand", copyCommand) - if _, err = s.ExecContext(ctx, copyCommand); err != nil { return fmt.Errorf("failed to run COPY INTO for temporary table: %w", err) } diff --git a/lib/typing/values/string.go b/lib/typing/values/string.go index 544be3a9e..8114ea610 100644 --- a/lib/typing/values/string.go +++ b/lib/typing/values/string.go @@ -97,6 +97,14 @@ func ToString(colVal any, colKind typing.KindDetails) (string, error) { } } case typing.Array.Kind: + // If the column value is TOASTED, we should return an array with the TOASTED placeholder + // We're doing this to make sure that the value matches the schema. + if stringValue, ok := colVal.(string); ok { + if stringValue == constants.ToastUnavailableValuePlaceholder { + return fmt.Sprintf(`["%s"]`, constants.ToastUnavailableValuePlaceholder), nil + } + } + colValBytes, err := json.Marshal(colVal) if err != nil { return "", err From 454d71c17b1ddeb2b027cd803158ef972b108db3 Mon Sep 17 00:00:00 2001 From: Robin Tang Date: Sun, 15 Dec 2024 22:16:53 -0800 Subject: [PATCH 3/4] Clean up. --- lib/typing/values/string_test.go | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/lib/typing/values/string_test.go b/lib/typing/values/string_test.go index 8db4751c4..5ad48a2af 100644 --- a/lib/typing/values/string_test.go +++ b/lib/typing/values/string_test.go @@ -172,9 +172,18 @@ func TestToString(t *testing.T) { } { // Array - val, err := ToString([]string{"foo", "bar"}, typing.Array) - assert.NoError(t, err) - assert.Equal(t, `["foo","bar"]`, val) + { + // Normal arrays + val, err := ToString([]string{"foo", "bar"}, typing.Array) + assert.NoError(t, err) + assert.Equal(t, `["foo","bar"]`, val) + } + { + // Toasted array + val, err := ToString(constants.ToastUnavailableValuePlaceholder, typing.Array) + assert.NoError(t, err) + assert.Equal(t, `["__debezium_unavailable_value"]`, val) + } } { // Integer column From a839e0d3c36a98ebd3b0a46a9723b21d2358865a Mon Sep 17 00:00:00 2001 From: Robin Tang Date: Sun, 15 Dec 2024 22:19:37 -0800 Subject: [PATCH 4/4] One line. --- clients/databricks/store.go | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/clients/databricks/store.go b/clients/databricks/store.go index 563912479..13d2b1618 100644 --- a/clients/databricks/store.go +++ b/clients/databricks/store.go @@ -149,10 +149,8 @@ FORMAT_OPTIONS ( );`, // COPY INTO tempTableID.FullyQualifiedName(), - // SELECT columns - strings.Join(ordinalColumns, ", "), - // FROM - file.DBFSFilePath(), + // SELECT columns FROM file + strings.Join(ordinalColumns, ", "), file.DBFSFilePath(), ) if _, err = s.ExecContext(ctx, copyCommand); err != nil {