feat: Implement date_part scalar function

cmd/influx/cli/cli.go

@@ -16,6 +16,7 @@ import (
 	"path/filepath"
 	"reflect"
 	"runtime"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -880,6 +881,9 @@ func headersEqual(prev, current models.Row) bool {
 	if prev.Name != current.Name {
 		return false
 	}
+	if !slices.Equal(prev.GroupingKeys, current.GroupingKeys) {
+		return false
+	}
 	return tagsEqual(prev.Tags, current.Tags) && columnsEqual(prev.Columns, current.Columns)
 }
 
@@ -890,9 +894,10 @@ func (c *CommandLine) writeCSV(response *client.Response, w io.Writer) {
 		suppressHeaders := len(result.Series) > 0 && headersEqual(previousHeaders, result.Series[0])
 		if !suppressHeaders && len(result.Series) > 0 {
 			previousHeaders = models.Row{
-				Name:    result.Series[0].Name,
-				Tags:    result.Series[0].Tags,
-				Columns: result.Series[0].Columns,
+				Name:         result.Series[0].Name,
+				Tags:         result.Series[0].Tags,
+				GroupingKeys: result.Series[0].GroupingKeys,
+				Columns:      result.Series[0].Columns,
 			}
 		}
 
@@ -920,9 +925,10 @@ func (c *CommandLine) writeColumns(response *client.Response, w io.Writer) {
 		suppressHeaders := len(result.Series) > 0 && headersEqual(previousHeaders, result.Series[0])
 		if !suppressHeaders && len(result.Series) > 0 {
 			previousHeaders = models.Row{
-				Name:    result.Series[0].Name,
-				Tags:    result.Series[0].Tags,
-				Columns: result.Series[0].Columns,
+				Name:         result.Series[0].Name,
+				Tags:         result.Series[0].Tags,
+				GroupingKeys: result.Series[0].GroupingKeys,
+				Columns:      result.Series[0].Columns,
 			}
 		}
 
@@ -984,6 +990,9 @@ func (c *CommandLine) formatResults(result client.Result, separator string, supp
 				t := fmt.Sprintf("tags: %s", (strings.Join(tags, ", ")))
 				rows = append(rows, t)
 			}
+			if len(row.GroupingKeys) > 0 {
+				rows = append(rows, fmt.Sprintf("group: %s", strings.Join(row.GroupingKeys, ", ")))
+			}
 		}
 
 		if !suppressHeaders {

coordinator/statement_executor.go

@@ -618,7 +618,7 @@ func (e *StatementExecutor) executeSelectStatement(ctx *query.ExecutionContext,
 			Messages: messages,
 			Series: []*models.Row{{
 				Name:    "result",
-				Columns: []string{"time", "written"},
+				Columns: []string{models.TimeString, "written"},
 				Values:  [][]interface{}{{time.Unix(0, 0).UTC(), writeN}},
 			}},
 		})
@@ -1394,13 +1394,34 @@ var errNoDatabaseInTarget = errors.New("no database in target")
 
 // convertRowToPoints will convert a query result Row into Points that can be written back in.
 func convertRowToPoints(measurementName string, row *models.Row, strictErrorHandling bool) ([]models.Point, error) {
-	// figure out which parts of the result are the time and which are the fields
+	// figure out which parts of the result are the time, the GROUP BY date_part
+	// grouping dimensions, and which are the regular fields.
+	//
+	// A GROUP BY date_part dimension (e.g. "year") is injected as an output column
+	// and recorded in row.GroupingKeys, but it identifies the series rather than
+	// carrying a value: every group shares the same representative bucket timestamp
+	// and the same base tag set (single window, Interval=0). Writing such a column
+	// as a field would collapse every group onto an identical series+timestamp
+	// (last-write-wins, silent data loss), so promote these columns to tags instead
+	// — mirroring how GROUP BY <tag> INTO writes its grouping tag.
+	var groupingCols map[string]struct{}
+	if len(row.GroupingKeys) > 0 {
+		groupingCols = make(map[string]struct{}, len(row.GroupingKeys))
+		for _, k := range row.GroupingKeys {
+			groupingCols[k] = struct{}{}
+		}
+	}
+
 	timeIndex := -1
 	fieldIndexes := make(map[string]int)
+	tagIndexes := make(map[string]int, len(row.GroupingKeys))
 	for i, c := range row.Columns {
-		if c == "time" {
+		switch {
+		case c == models.TimeString:
 			timeIndex = i
-		} else {
+		case isGroupingColumn(groupingCols, c):
+			tagIndexes[c] = i
+		default:
 			fieldIndexes[c] = i
 		}
 	}
@@ -1423,7 +1444,7 @@ func convertRowToPoints(measurementName string, row *models.Row, strictErrorHand
 			}
 		}
 
-		p, err := models.NewPoint(measurementName, models.NewTags(row.Tags), vals, v[timeIndex].(time.Time))
+		p, err := models.NewPoint(measurementName, groupingTags(row.Tags, tagIndexes, v), vals, v[timeIndex].(time.Time))
 		if err != nil {
 			if !strictErrorHandling {
 				// Drop points that can't be stored
@@ -1438,6 +1459,55 @@ func convertRowToPoints(measurementName string, row *models.Row, strictErrorHand
 	return points, nil
 }
 
+// isGroupingColumn reports whether column c is a GROUP BY date_part grouping
+// dimension (and so must be written as a tag rather than a field).
+func isGroupingColumn(groupingCols map[string]struct{}, c string) bool {
+	_, ok := groupingCols[c]
+	return ok
+}
+
+// groupingTags builds the tag set for a single INTO point. It starts from the
+// row's base tags (shared by every group) and adds the per-group date_part
+// dimension values from tagIndexes so each group becomes a distinct series. When
+// there are no grouping dimensions this is equivalent to models.NewTags(base).
+func groupingTags(base map[string]string, tagIndexes map[string]int, v []interface{}) models.Tags {
+	if len(tagIndexes) == 0 {
+		return models.NewTags(base)
+	}
+
+	merged := make(map[string]string, len(base)+len(tagIndexes))
+	for k, val := range base {
+		merged[k] = val
+	}
+	for tagName, tagIndex := range tagIndexes {
+		if s, ok := tagValueString(v[tagIndex]); ok {
+			merged[tagName] = s
+		}
+	}
+	return models.NewTags(merged)
+}
+
+// tagValueString renders a date_part dimension value as a tag value. date_part
+// dimensions are emitted as int64 (influxql.Integer); other types are rendered
+// defensively. A nil value yields ok=false so the tag is omitted rather than
+// written as an empty (and therefore dropped) tag.
+func tagValueString(val interface{}) (string, bool) {
+	switch x := val.(type) {
+	case nil:
+		return "", false
+	case int64:
+		return strconv.FormatInt(x, 10), true
+	case int:
+		return strconv.FormatInt(int64(x), 10), true
+	case int32:
+		return strconv.FormatInt(int64(x), 10), true
+	case string:
+		return x, x != ""
+	default:
+		return fmt.Sprintf("%v", x), true
+	}
+}
+
 // NormalizeStatement adds a default database and policy to the measurements in statement.
 // Parameter defaultRetentionPolicy can be "".
 func (e *StatementExecutor) NormalizeStatement(stmt influxql.Statement, defaultDatabase, defaultRetentionPolicy string) (err error) {

coordinator/statement_executor_test.go

@@ -710,9 +710,9 @@ func TestStatementExecutor_ExecuteShowMeasurementsStatement_Partial(t *testing.T
 		// Source labels match influxql.QuoteIdent: simple names emit unquoted,
 		// joined by '.'. Update these alongside formatMeasurementSource if the
 		// labeling format changes.
-		quotedDB0     = DefaultDatabase
-		quotedDB0RP0  = DefaultDatabase + "." + DefaultRetentionPolicy
-		quotedDB1RP0  = db1Name + "." + DefaultRetentionPolicy
+		quotedDB0    = DefaultDatabase
+		quotedDB0RP0 = DefaultDatabase + "." + DefaultRetentionPolicy
+		quotedDB1RP0 = db1Name + "." + DefaultRetentionPolicy
 	)
 
 	wildcardDBs := func() []meta.DatabaseInfo {

models/rows.go

@@ -6,16 +6,20 @@ import (
 
 // Row represents a single row returned from the execution of a statement.
 type Row struct {
-	Name    string            `json:"name,omitempty"`
-	Tags    map[string]string `json:"tags,omitempty"`
-	Columns []string          `json:"columns,omitempty"`
-	Values  [][]interface{}   `json:"values,omitempty"`
-	Partial bool              `json:"partial,omitempty"`
+	Name         string            `json:"name,omitempty"`
+	Tags         map[string]string `json:"tags,omitempty"`
+	GroupingKeys []string          `json:"grouping_keys,omitempty"`
+	Columns      []string          `json:"columns,omitempty"`
+	Values       [][]interface{}   `json:"values,omitempty"`
+	Partial      bool              `json:"partial,omitempty"`
 }
 
 // SameSeries returns true if r contains values for the same series as o.
 func (r *Row) SameSeries(o *Row) bool {
-	return r.tagsHash() == o.tagsHash() && r.Name == o.Name
+	if len(o.GroupingKeys) == 0 && len(r.GroupingKeys) == 0 {
+		return r.tagsHash() == o.tagsHash() && r.Name == o.Name
+	}
+	return r.tagsHash() == o.tagsHash() && r.Name == o.Name && r.groupingKeysHash() == o.groupingKeysHash()
 }
 
 // tagsHash returns a hash of tag key/value pairs.
@@ -29,6 +33,17 @@ func (r *Row) tagsHash() uint64 {
 	return h.Sum64()
 }
 
+func (r *Row) groupingKeysHash() uint64 {
+	h := NewInlineFNV64a()
+	for i, k := range r.GroupingKeys {
+		if i > 0 {
+			h.Write([]byte{0})
+		}
+		h.Write([]byte(k))
+	}
+	return h.Sum64()
+}
+
 // tagKeys returns a sorted list of tag keys.
 func (r *Row) tagsKeys() []string {
 	a := make([]string, 0, len(r.Tags))

models/time.go

@@ -39,6 +39,12 @@ var (
 
 	// ErrTimeOutOfRange gets returned when time is out of the representable range using int64 nanoseconds since the epoch.
 	ErrTimeOutOfRange = fmt.Errorf("time outside range %d - %d", MinNanoTime, MaxNanoTime)
+
+	// Static objects to prevent small allocs.
+	TimeBytes = []byte("time")
+
+	// TimeString is a variable representing the string "time"
+	TimeString = string(TimeBytes)
 )
 
 // SafeCalcTime safely calculates the time given. Will return error if the time is outside the