feat: Implement date_part scalar function

cmd/influx/cli/cli.go

@@ -880,6 +880,9 @@ func headersEqual(prev, current models.Row) bool {
 	if prev.Name != current.Name {
 		return false
 	}
+	if !reflect.DeepEqual(prev.GroupingKeys, current.GroupingKeys) {
+		return false
+	}
 	return tagsEqual(prev.Tags, current.Tags) && columnsEqual(prev.Columns, current.Columns)
 }
 
@@ -890,9 +893,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 +924,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 +989,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,31 @@ 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.
+	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)
 	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 +1441,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 +1456,58 @@ 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 {
+	if len(groupingCols) == 0 {
+		return false
+	}
+	_, 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) {

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