enhance: [2.6] Show create time for import job (#45059)

issue: #45056

pr: #45058

---------

Signed-off-by: bigsheeper <[email protected]>

Author: bigsheeper

internal/datacoord/import_job.go

@@ -118,7 +118,7 @@ type ImportJob interface {
 	GetState() internalpb.ImportJobState
 	GetReason() string
 	GetRequestedDiskSize() int64
-	GetStartTime() string
+	GetCreateTime() string
 	GetCompleteTime() string
 	GetFiles() []*internalpb.ImportFile
 	GetOptions() []*commonpb.KeyValuePair

internal/datacoord/services.go

@@ -1898,7 +1898,7 @@ func (s *Server) ImportV2(ctx context.Context, in *internalpb.ImportRequestInter
 	if jobID == 0 {
 		jobID = idStart
 	}
-	startTime := time.Now()
+	createTime := time.Now()
 	job := &importJob{
 		ImportJob: &datapb.ImportJob{
 			JobID:          jobID,
@@ -1912,7 +1912,7 @@ func (s *Server) ImportV2(ctx context.Context, in *internalpb.ImportRequestInter
 			State:          internalpb.ImportJobState_Pending,
 			Files:          files,
 			Options:        in.GetOptions(),
-			StartTime:      startTime.Format("2006-01-02T15:04:05Z07:00"),
+			CreateTime:     createTime.Format("2006-01-02T15:04:05Z07:00"),
 			ReadyVchannels: in.GetChannelNames(),
 			DataTs:         in.GetDataTimestamp(),
 		},
@@ -1961,7 +1961,7 @@ func (s *Server) GetImportProgress(ctx context.Context, in *internalpb.GetImport
 	resp.Reason = reason
 	resp.Progress = progress
 	resp.CollectionName = job.GetCollectionName()
-	resp.StartTime = job.GetStartTime()
+	resp.CreateTime = job.GetCreateTime()
 	resp.CompleteTime = job.GetCompleteTime()
 	resp.ImportedRows = importedRows
 	resp.TotalRows = totalRows

internal/distributed/proxy/httpserver/handler_v2.go

@@ -2737,6 +2737,7 @@ func (h *HandlersV2) getImportJobProcess(ctx context.Context, c *gin.Context, an
 		response := resp.(*internalpb.GetImportProgressResponse)
 		returnData := make(map[string]interface{})
 		returnData["jobId"] = jobIDGetter.GetJobID()
+		returnData["createTime"] = response.GetCreateTime()
 		returnData["collectionName"] = response.GetCollectionName()
 		returnData["completeTime"] = response.GetCompleteTime()
 		returnData["state"] = response.GetState().String()

internal/proxy/impl.go

@@ -4602,7 +4602,7 @@ func convertToV1GetImportResponse(rsp *internalpb.GetImportProgressResponse) *mi
 		Value: strconv.FormatInt(rsp.GetProgress(), 10),
 	})
 	var createTs int64
-	createTime, err := time.Parse("2006-01-02T15:04:05Z07:00", rsp.GetStartTime())
+	createTime, err := time.Parse("2006-01-02T15:04:05Z07:00", rsp.GetCreateTime())
 	if err == nil {
 		createTs = createTime.Unix()
 	}

internal/util/importutilv2/common/util.go

@@ -18,6 +18,8 @@ package common
 
 import (
 	"fmt"
+	"strings"
+	"unicode/utf8"
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
@@ -55,9 +57,41 @@ func EstimateReadCountPerBatch(bufferSize int, schema *schemapb.CollectionSchema
 	return ret, nil
 }
 
+// SafeStringForError safely converts a string for use in error messages.
+// It replaces invalid UTF-8 sequences with their hex representation to avoid
+// gRPC serialization errors while still providing useful debugging information.
+func SafeStringForError(s string) string {
+	if utf8.ValidString(s) {
+		return s
+	}
+
+	var result strings.Builder
+	for i, r := range s {
+		if r == utf8.RuneError {
+			// Invalid UTF-8 sequence, encode as hex
+			result.WriteString(fmt.Sprintf("\\x%02x", s[i]))
+		} else {
+			result.WriteRune(r)
+		}
+	}
+	return result.String()
+}
+
+// SafeStringForErrorWithLimit safely converts a string for use in error messages
+// with a length limit to prevent extremely long error messages.
+func SafeStringForErrorWithLimit(s string, maxLen int) string {
+	safe := SafeStringForError(s)
+	if len(safe) <= maxLen {
+		return safe
+	}
+	return safe[:maxLen] + "..."
+}
+
 func CheckValidUTF8(s string, field *schemapb.FieldSchema) error {
 	if !typeutil.IsUTF8(s) {
-		return fmt.Errorf("field %s contains invalid UTF-8 data, value=%s", field.GetName(), s)
+		// Use safe string representation to avoid gRPC serialization errors
+		safeValue := SafeStringForErrorWithLimit(s, 100)
+		return fmt.Errorf("field '%s' contains invalid UTF-8 data, value=%s", field.GetName(), safeValue)
 	}
 	return nil
 }

internal/util/importutilv2/common/util_test.go

@@ -17,7 +17,9 @@
 package common
 
 import (
+	"strings"
 	"testing"
+	"unicode/utf8"
 
 	"github.com/stretchr/testify/assert"
 
@@ -181,3 +183,81 @@ func TestUtil_CheckValidString(t *testing.T) {
 	err = CheckValidString("aaaaa", 5, fieldSchema)
 	assert.NoError(t, err)
 }
+
+func TestUtil_SafeStringForError(t *testing.T) {
+	// Test valid UTF-8 string
+	validStr := "Hello, 世界!"
+	result := SafeStringForError(validStr)
+	assert.Equal(t, validStr, result)
+
+	// Test invalid UTF-8 string
+	invalidStr := string([]byte{0xC0, 0xAF, 'a', 'b', 'c'})
+	result = SafeStringForError(invalidStr)
+	assert.Contains(t, result, "\\xc0")
+	assert.Contains(t, result, "\\xaf")
+	assert.Contains(t, result, "abc")
+
+	// Test empty string
+	result = SafeStringForError("")
+	assert.Equal(t, "", result)
+
+	// Test string with mixed valid and invalid UTF-8
+	mixedStr := "valid" + string([]byte{0xFF, 0xFE}) + "text"
+	result = SafeStringForError(mixedStr)
+	assert.Contains(t, result, "valid")
+	assert.Contains(t, result, "\\xff")
+	assert.Contains(t, result, "\\xfe")
+	assert.Contains(t, result, "text")
+}
+
+func TestUtil_SafeStringForErrorWithLimit(t *testing.T) {
+	// Test string within limit
+	shortStr := "short"
+	result := SafeStringForErrorWithLimit(shortStr, 10)
+	assert.Equal(t, shortStr, result)
+
+	// Test string exceeding limit
+	longStr := "this is a very long string that exceeds the limit"
+	result = SafeStringForErrorWithLimit(longStr, 20)
+	assert.Equal(t, 23, len(result)) // 20 chars + "..."
+	assert.True(t, strings.HasSuffix(result, "..."))
+
+	// Test invalid UTF-8 string with limit
+	invalidStr := string([]byte{0xC0, 0xAF, 0xFF, 0xFE, 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z'})
+	result = SafeStringForErrorWithLimit(invalidStr, 15)
+	assert.True(t, len(result) <= 18) // 15 chars + "..."
+	assert.True(t, strings.HasSuffix(result, "..."))
+}
+
+func TestUtil_CheckValidUTF8_WithSafeError(t *testing.T) {
+	fieldSchema := &schemapb.FieldSchema{
+		FieldID:  1,
+		Name:     "test_field",
+		DataType: schemapb.DataType_VarChar,
+		TypeParams: []*commonpb.KeyValuePair{
+			{
+				Key:   common.MaxLengthKey,
+				Value: "1000",
+			},
+		},
+	}
+
+	// Test with invalid UTF-8 - should not cause gRPC serialization error
+	invalidStr := string([]byte{0xC0, 0xAF, 0xFF, 0xFE})
+	err := CheckValidUTF8(invalidStr, fieldSchema)
+	assert.Error(t, err)
+
+	// Verify the error message contains safe representation
+	errMsg := err.Error()
+	assert.Contains(t, errMsg, "test_field")
+	assert.Contains(t, errMsg, "invalid UTF-8 data")
+	assert.Contains(t, errMsg, "\\xc0") // Should contain hex representation
+	assert.Contains(t, errMsg, "\\xaf")
+
+	// Verify the error message is valid UTF-8 itself
+	assert.True(t, utf8.ValidString(errMsg), "Error message should be valid UTF-8")
+
+	// Test with valid UTF-8
+	err = CheckValidUTF8("valid string", fieldSchema)
+	assert.NoError(t, err)
+}

pkg/proto/data_coord.proto

@@ -987,7 +987,7 @@ message ImportJob {
   string complete_time = 13;
   repeated internal.ImportFile files = 14;
   repeated common.KeyValuePair options = 15;
-  string start_time = 16;
+  string create_time = 16;
   repeated string ready_vchannels = 17;
   uint64 data_ts = 18;
 }