diff --git a/clients/databricks/store.go b/clients/databricks/store.go index 02df68b9..1056e8dc 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) }