diff --git a/runtime/drivers/clickhouse/dialect.go b/runtime/drivers/clickhouse/dialect.go index 8ee1dc947f7..845710853eb 100644 --- a/runtime/drivers/clickhouse/dialect.go +++ b/runtime/drivers/clickhouse/dialect.go @@ -188,7 +188,7 @@ func (d *dialect) SelectTimeRangeBins(start, end time.Time, grain runtimev1.Time func (d *dialect) SelectInlineResults(result *drivers.Result) (string, []any, []any, error) { for _, f := range result.Schema.Fields { - if !drivers.CheckTypeCompatibility(f) { + if !inlineTypeSupported(f) { return "", nil, nil, fmt.Errorf("select inline: schema field type not supported %q: %w", f.Type.Code, drivers.ErrOptimizationFailure) } } @@ -231,7 +231,7 @@ func (d *dialect) SelectInlineResults(result *drivers.Result) (string, []any, [] prefix += ", " } } - argExpr, argVal, err := getArgExpr(v, result.Schema.Fields[i].Type.Code) + argExpr, argVal, err := getArgExpr(v, result.Schema.Fields[i].Type) if err != nil { return "", nil, nil, fmt.Errorf("select inline: failed to get argument expression: %w", err) } @@ -300,16 +300,40 @@ func (d *dialect) ColumnNumericHistogramBucket(db, dbSchema, table, column strin d.EscapeTable(db, dbSchema, table)), nil } -func getArgExpr(val any, typ runtimev1.Type_Code) (string, any, error) { +// inlineTypeSupported reports whether values of the field's type can be inlined by SelectInlineResults. +// Besides the types that the driver binds natively, it allows UUID and Decimal values, which getArgExpr casts back to their ClickHouse type. +func inlineTypeSupported(f *runtimev1.StructType_Field) bool { + if drivers.CheckTypeCompatibility(f) { + return true + } + switch f.Type.Code { + case runtimev1.Type_CODE_UUID, runtimev1.Type_CODE_DECIMAL: + // The cast needs the exact ClickHouse type name, which is only known for the results of a ClickHouse query. + return f.Type.RawType != "" + default: + return false + } +} + +// getArgExpr returns the SQL expression and the positional argument for inlining val in a VALUES clause. +func getArgExpr(val any, typ *runtimev1.Type) (string, any, error) { if val == nil { return "NULL", nil, nil } - if typ == runtimev1.Type_CODE_DATE { + switch typ.Code { + case runtimev1.Type_CODE_DATE: + // Dates are formatted explicitly, otherwise time.Time args are bound as DateTime values. t, ok := val.(time.Time) if !ok { return "", nil, fmt.Errorf("could not cast value %v to time.Time for date type", val) } return "toDate(?)", t.Format(time.DateOnly), nil + case runtimev1.Type_CODE_UUID, runtimev1.Type_CODE_DECIMAL: + // The driver binds UUID and Decimal values as quoted strings, so VALUES would infer a String column. + // A String column can neither be joined with nor compared to the original UUID or Decimal column, + // so the value is cast back to its exact ClickHouse type, e.g. "Nullable(UUID)" or "Decimal(38, 4)". + return fmt.Sprintf("CAST(? AS %s)", typ.RawType), fmt.Sprint(val), nil + default: + return "?", val, nil } - return "?", val, nil } diff --git a/runtime/drivers/clickhouse/olap_test.go b/runtime/drivers/clickhouse/olap_test.go index 7b1048bba31..0f5d5913064 100644 --- a/runtime/drivers/clickhouse/olap_test.go +++ b/runtime/drivers/clickhouse/olap_test.go @@ -47,6 +47,7 @@ func TestClickhouseSingle(t *testing.T) { t.Run("QueryAttributesReadonlyUser", func(t *testing.T) { testQueryAttributesReadonlyUser(t, olap, dsn) }) t.Run("CreateTableAsSelect_WithPrePostExec", func(t *testing.T) { testCreateTableAsSelect_WithPrePostExec(t, c, olap) }) t.Run("InsertTableAsSelect_WithPrePostExec", func(t *testing.T) { testInsertTableAsSelect_WithPrePostExec(t, c, olap) }) + t.Run("SelectInlineResults", func(t *testing.T) { testSelectInlineResults(t, olap) }) } func TestClickhouseCluster(t *testing.T) { @@ -1220,3 +1221,61 @@ func testEntityTypeRestrictedUser(t *testing.T, olap drivers.OLAPStore, dsn, clu require.NoError(t, err) require.False(t, exists) } + +// testSelectInlineResults checks that query results can be inlined in a VALUES clause that preserves their types. +// UUID and Decimal values are bound as strings, so they must be cast back to their ClickHouse type to stay joinable with the original columns. +func testSelectInlineResults(t *testing.T, olap drivers.OLAPStore) { + ctx := context.Background() + res, err := olap.Query(ctx, &drivers.Statement{ + Query: `SELECT * FROM ( + SELECT toUUID('6ba7b810-9dad-11d1-80b4-00c04fd430c8') AS id, toNullable(toUUID('6ba7b811-9dad-11d1-80b4-00c04fd430c8')) AS nid, toDecimal64(1.25, 2) AS price, toNullable(toDecimal128(-3.5, 4)) AS nprice, 'a' AS name, toDate('2024-01-02') AS day + UNION ALL + SELECT toUUID('6ba7b812-9dad-11d1-80b4-00c04fd430c8'), NULL, toDecimal64(0, 2), NULL, 'b', toDate('2024-01-03') + ) ORDER BY name`, + }) + require.NoError(t, err) + defer res.Close() + + sql, args, dimVals, err := olap.Dialect().SelectInlineResults(res) + require.NoError(t, err) + require.Len(t, dimVals, 2) + require.Contains(t, sql, "CAST(? AS UUID)") + require.Contains(t, sql, "CAST(? AS Nullable(UUID))") + require.Contains(t, sql, "CAST(? AS Decimal(18, 2))") + require.Contains(t, sql, "CAST(? AS Nullable(Decimal(38, 4)))") + require.Contains(t, sql, "toDate(?)") + + // Round-trip the inlined results through ClickHouse and check the values and the types they come back with. + rows, err := olap.Query(ctx, &drivers.Statement{ + Query: fmt.Sprintf(`SELECT toTypeName(id) AS id_type, toString(id) AS id_text, toTypeName(nid) AS nid_type, ifNull(toString(nid), '') AS nid_text, toTypeName(price) AS price_type, toString(price) AS price_text, toTypeName(nprice) AS nprice_type, ifNull(toString(nprice), '') AS nprice_text, name, toTypeName(day) AS day_type, toString(day) AS day_text FROM (%s) ORDER BY name`, sql), + Args: args, + }) + require.NoError(t, err) + defer rows.Close() + + var got []map[string]any + for rows.Next() { + row := make(map[string]any) + require.NoError(t, rows.MapScan(row)) + got = append(got, row) + } + require.NoError(t, rows.Err()) + require.Equal(t, []map[string]any{ + { + "id_type": "UUID", "id_text": "6ba7b810-9dad-11d1-80b4-00c04fd430c8", + "nid_type": "Nullable(UUID)", "nid_text": "6ba7b811-9dad-11d1-80b4-00c04fd430c8", + "price_type": "Decimal(18, 2)", "price_text": "1.25", + "nprice_type": "Nullable(Decimal(38, 4))", "nprice_text": "-3.5", + "name": "a", + "day_type": "Date", "day_text": "2024-01-02", + }, + { + "id_type": "UUID", "id_text": "6ba7b812-9dad-11d1-80b4-00c04fd430c8", + "nid_type": "Nullable(UUID)", "nid_text": "", + "price_type": "Decimal(18, 2)", "price_text": "0", + "nprice_type": "Nullable(Decimal(38, 4))", "nprice_text": "", + "name": "b", + "day_type": "Date", "day_text": "2024-01-03", + }, + }, got) +} diff --git a/runtime/metricsview/executor/executor_rewrite_two_phase_comparisons_integration_test.go b/runtime/metricsview/executor/executor_rewrite_two_phase_comparisons_integration_test.go new file mode 100644 index 00000000000..de112a76731 --- /dev/null +++ b/runtime/metricsview/executor/executor_rewrite_two_phase_comparisons_integration_test.go @@ -0,0 +1,127 @@ +package executor_test + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/rilldata/rill/runtime" + "github.com/rilldata/rill/runtime/drivers" + "github.com/rilldata/rill/runtime/metricsview" + "github.com/rilldata/rill/runtime/metricsview/executor" + "github.com/rilldata/rill/runtime/testruntime" + "github.com/stretchr/testify/require" +) + +// A comparison leaderboard on a UUID dimension with a Decimal measure must take the two-phase path: +// the base results are inlined with a cast back to their ClickHouse types, which the second phase joins on and filters the comparison range with. +// Before the cast, UUID and Decimal results were rejected as not inlinable and the query silently aggregated the whole comparison range instead. +func TestRewriteTwoPhaseComparisonsUUIDDimension(t *testing.T) { + rt, instanceID := testruntime.NewInstanceWithOptions(t, testruntime.InstanceOptions{ + TestConnectors: []string{"clickhouse"}, + Files: map[string]string{ + "rill.yaml": "olap_connector: clickhouse", + // 20 users with five events on each of ten days; users 9 and 19 have a NULL id, and the price makes every user's revenue distinct. + "events.sql": ` +-- @connector: clickhouse +select + toDateTime('2024-01-01 00:00:00') + interval (number % 10) day as event_time, + if(intDiv(number, 10) % 10 = 9, NULL, toUUID(UUIDNumToString(MD5(toString(intDiv(number, 10) % 20))))) as user_id, + toDecimal64(intDiv(number, 10) % 20 + 0.25 + (number % 10) * 0.1, 2) as price +from numbers(1000) +`, + "mv.yaml": ` +type: metrics_view +version: 1 +model: events +timeseries: event_time +dimensions: + - name: user_id + column: user_id +measures: + - name: events + expression: count(*) + - name: revenue + expression: sum(price) +`, + }, + }) + testruntime.RequireReconcileState(t, rt, instanceID, 3, 0, 0) + + r := testruntime.GetResource(t, rt, instanceID, runtime.ResourceKindMetricsView, "mv") + mv := r.GetMetricsView().State.ValidSpec + require.NotNil(t, mv) + + ctx := context.Background() + olap, release, err := rt.OLAP(ctx, instanceID, mv.Connector) + require.NoError(t, err) + defer release() + + e, err := executor.New(ctx, rt, instanceID, mv, false, runtime.ResolvedSecurityOpen, 0, nil) + require.NoError(t, err) + defer e.Close() + + limit := int64(5) + qry := &metricsview.Query{ + Dimensions: []metricsview.Dimension{{Name: "user_id"}}, + Measures: []metricsview.Measure{ + {Name: "events"}, + {Name: "revenue"}, + {Name: "events__prev", Compute: &metricsview.MeasureCompute{ComparisonValue: &metricsview.MeasureComputeComparisonValue{Measure: "events"}}}, + {Name: "revenue__prev", Compute: &metricsview.MeasureCompute{ComparisonValue: &metricsview.MeasureComputeComparisonValue{Measure: "revenue"}}}, + {Name: "revenue__delta", Compute: &metricsview.MeasureCompute{ComparisonDelta: &metricsview.MeasureComputeComparisonDelta{Measure: "revenue"}}}, + }, + Sort: []metricsview.Sort{{Name: "revenue", Desc: true}}, + TimeRange: &metricsview.TimeRange{Start: time.Date(2024, 1, 6, 0, 0, 0, 0, time.UTC), End: time.Date(2024, 1, 11, 0, 0, 0, 0, time.UTC)}, + ComparisonTimeRange: &metricsview.TimeRange{Start: time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC), End: time.Date(2024, 1, 6, 0, 0, 0, 0, time.UTC)}, + Limit: &limit, + } + + res, err := e.Query(ctx, qry, nil) + require.NoError(t, err) + got := twoPhaseTestResultRows(t, res) + + // The same leaderboard as a single query, with the limit applied to the base range only. + table := olap.Dialect().EscapeTable(mv.Database, mv.DatabaseSchema, mv.Table) + want := twoPhaseTestRows(t, olap, fmt.Sprintf(` + SELECT b.user_id AS user_id, b.events AS events, b.revenue AS revenue, c.events AS events__prev, c.revenue AS revenue__prev, b.revenue - c.revenue AS revenue__delta + FROM (SELECT user_id, count(*) AS events, sum(price) AS revenue FROM %[1]s WHERE event_time >= toDateTime('2024-01-06 00:00:00') AND event_time < toDateTime('2024-01-11 00:00:00') GROUP BY user_id ORDER BY revenue DESC LIMIT 5) b + LEFT JOIN (SELECT user_id, count(*) AS events, sum(price) AS revenue FROM %[1]s WHERE event_time >= toDateTime('2024-01-01 00:00:00') AND event_time < toDateTime('2024-01-06 00:00:00') GROUP BY user_id) c + ON isNotDistinctFrom(b.user_id, c.user_id) + ORDER BY revenue DESC`, table), nil) + require.Len(t, want, 5) + require.Equal(t, "", want[0]["user_id"], "the NULL user must be in the leaderboard so that the rewrite's IS NULL branch is exercised") + require.Equal(t, want, got) + + // The second phase must have run with the base results inlined, cast back to the dimension's and measure's types. + err = olap.Exec(ctx, &drivers.Statement{Query: "SYSTEM FLUSH LOGS"}) + require.NoError(t, err) + logged := twoPhaseTestRows(t, olap, "SELECT count() AS n FROM system.query_log WHERE type = 'QueryFinish' AND query LIKE '%FROM VALUES%' AND query LIKE '%CAST(%AS Nullable(UUID))%' AND query LIKE '%CAST(%AS Decimal(38, 2))%'", nil) + require.Equal(t, "1", logged[0]["n"]) +} + +func twoPhaseTestRows(t *testing.T, olap drivers.OLAPStore, sql string, args []any) []map[string]string { + t.Helper() + res, err := olap.Query(context.Background(), &drivers.Statement{Query: sql, Args: args}) + require.NoError(t, err) + return twoPhaseTestResultRows(t, res) +} + +// twoPhaseTestResultRows returns the rows with every value formatted as a string, because VALUES infers the narrowest integer type for inlined counts. +func twoPhaseTestResultRows(t *testing.T, res *drivers.Result) []map[string]string { + t.Helper() + defer res.Close() + var rows []map[string]string + for res.Next() { + row := make(map[string]any) + require.NoError(t, res.MapScan(row)) + formatted := make(map[string]string, len(row)) + for k, v := range row { + formatted[k] = fmt.Sprint(v) + } + rows = append(rows, formatted) + } + require.NoError(t, res.Err()) + return rows +}