fix: inline UUID and Decimal results in two-phase comparison queries on ClickHouse

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
 }

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), '<null>') AS nid_text, toTypeName(price) AS price_type, toString(price) AS price_text, toTypeName(nprice) AS nprice_type, ifNull(toString(nprice), '<null>') 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": "<null>",
+			"price_type": "Decimal(18, 2)", "price_text": "0",
+			"nprice_type": "Nullable(Decimal(38, 4))", "nprice_text": "<null>",
+			"name":     "b",
+			"day_type": "Date", "day_text": "2024-01-03",
+		},
+	}, got)
+}

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, "<nil>", 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
+}