Mirror
/
seaweedfs
mirror of https://github.com/seaweedfs/seaweedfs.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

rename

mq-subscribe
chrislu 8 months ago
parent
111a4e1126
commit
60103171b6
2 changed files with 22 additions and 22 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 32
      weed/mq/schema/value_builder.go
  2. 12
      weed/mq/schema/write_parquet_test.go

32
weed/mq/schema/value_builder.go
View File

@ -17,35 +17,35 @@ func (rvb *RecordValueBuilder) Build() *schema_pb.RecordValue {
return rvb.recordValue return rvb.recordValue
} }
func (rvb *RecordValueBuilder) AddBoolValue(key string, value bool) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetBoolValue(key string, value bool) *RecordValueBuilder {
rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_BoolValue{BoolValue: value}} rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_BoolValue{BoolValue: value}}
return rvb return rvb
} }
func (rvb *RecordValueBuilder) AddIntValue(key string, value int32) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetIntValue(key string, value int32) *RecordValueBuilder {
rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_Int32Value{Int32Value: value}} rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_Int32Value{Int32Value: value}}
return rvb return rvb
} }
func (rvb *RecordValueBuilder) AddLongValue(key string, value int64) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetLongValue(key string, value int64) *RecordValueBuilder {
rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_Int64Value{Int64Value: value}} rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_Int64Value{Int64Value: value}}
return rvb return rvb
} }
func (rvb *RecordValueBuilder) AddFloatValue(key string, value float32) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetFloatValue(key string, value float32) *RecordValueBuilder {
rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_FloatValue{FloatValue: value}} rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_FloatValue{FloatValue: value}}
return rvb return rvb
} }
func (rvb *RecordValueBuilder) AddDoubleValue(key string, value float64) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetDoubleValue(key string, value float64) *RecordValueBuilder {
rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_DoubleValue{DoubleValue: value}} rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_DoubleValue{DoubleValue: value}}
return rvb return rvb
} }
func (rvb *RecordValueBuilder) AddBytesValue(key string, value []byte) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetBytesValue(key string, value []byte) *RecordValueBuilder {
rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_BytesValue{BytesValue: value}} rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_BytesValue{BytesValue: value}}
return rvb return rvb
} }
func (rvb *RecordValueBuilder) AddStringValue(key string, value string) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetStringValue(key string, value string) *RecordValueBuilder {
rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_StringValue{StringValue: value}} rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_StringValue{StringValue: value}}
return rvb return rvb
} }
func (rvb *RecordValueBuilder) AddRecordValue(key string, value *RecordValueBuilder) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetRecordValue(key string, value *RecordValueBuilder) *RecordValueBuilder {
rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_RecordValue{RecordValue: value.Build()}} rvb.recordValue.Fields[key] = &schema_pb.Value{Kind: &schema_pb.Value_RecordValue{RecordValue: value.Build()}}
return rvb return rvb
} }
@ -55,56 +55,56 @@ func (rvb *RecordValueBuilder) addListValue(key string, values []*schema_pb.Valu
return rvb return rvb
} }
func (rvb *RecordValueBuilder) AddBoolListValue(key string, values ...bool) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetBoolListValue(key string, values ...bool) *RecordValueBuilder {
var listValues []*schema_pb.Value var listValues []*schema_pb.Value
for _, v := range values { for _, v := range values {
listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_BoolValue{BoolValue: v}}) listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_BoolValue{BoolValue: v}})
} }
return rvb.addListValue(key, listValues) return rvb.addListValue(key, listValues)
} }
func (rvb *RecordValueBuilder) AddIntListValue(key string, values ...int32) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetIntListValue(key string, values ...int32) *RecordValueBuilder {
var listValues []*schema_pb.Value var listValues []*schema_pb.Value
for _, v := range values { for _, v := range values {
listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_Int32Value{Int32Value: v}}) listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_Int32Value{Int32Value: v}})
} }
return rvb.addListValue(key, listValues) return rvb.addListValue(key, listValues)
} }
func (rvb *RecordValueBuilder) AddLongListValue(key string, values ...int64) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetLongListValue(key string, values ...int64) *RecordValueBuilder {
var listValues []*schema_pb.Value var listValues []*schema_pb.Value
for _, v := range values { for _, v := range values {
listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_Int64Value{Int64Value: v}}) listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_Int64Value{Int64Value: v}})
} }
return rvb.addListValue(key, listValues) return rvb.addListValue(key, listValues)
} }
func (rvb *RecordValueBuilder) AddFloatListValue(key string, values ...float32) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetFloatListValue(key string, values ...float32) *RecordValueBuilder {
var listValues []*schema_pb.Value var listValues []*schema_pb.Value
for _, v := range values { for _, v := range values {
listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_FloatValue{FloatValue: v}}) listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_FloatValue{FloatValue: v}})
} }
return rvb.addListValue(key, listValues) return rvb.addListValue(key, listValues)
} }
func (rvb *RecordValueBuilder) AddDoubleListValue(key string, values ...float64) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetDoubleListValue(key string, values ...float64) *RecordValueBuilder {
var listValues []*schema_pb.Value var listValues []*schema_pb.Value
for _, v := range values { for _, v := range values {
listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_DoubleValue{DoubleValue: v}}) listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_DoubleValue{DoubleValue: v}})
} }
return rvb.addListValue(key, listValues) return rvb.addListValue(key, listValues)
} }
func (rvb *RecordValueBuilder) AddBytesListValue(key string, values ...[]byte) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetBytesListValue(key string, values ...[]byte) *RecordValueBuilder {
var listValues []*schema_pb.Value var listValues []*schema_pb.Value
for _, v := range values { for _, v := range values {
listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_BytesValue{BytesValue: v}}) listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_BytesValue{BytesValue: v}})
} }
return rvb.addListValue(key, listValues) return rvb.addListValue(key, listValues)
} }
func (rvb *RecordValueBuilder) AddStringListValue(key string, values ...string) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetStringListValue(key string, values ...string) *RecordValueBuilder {
var listValues []*schema_pb.Value var listValues []*schema_pb.Value
for _, v := range values { for _, v := range values {
listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_StringValue{StringValue: v}}) listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_StringValue{StringValue: v}})
} }
return rvb.addListValue(key, listValues) return rvb.addListValue(key, listValues)
} }
func (rvb *RecordValueBuilder) AddRecordListValue(key string, values ...*RecordValueBuilder) *RecordValueBuilder {
func (rvb *RecordValueBuilder) SetRecordListValue(key string, values ...*RecordValueBuilder) *RecordValueBuilder {
var listValues []*schema_pb.Value var listValues []*schema_pb.Value
for _, v := range values { for _, v := range values {
listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_RecordValue{RecordValue: v.Build()}}) listValues = append(listValues, &schema_pb.Value{Kind: &schema_pb.Value_RecordValue{RecordValue: v.Build()}})

12
weed/mq/schema/write_parquet_test.go
View File

@ -69,17 +69,17 @@ func testWritingParquetFile(t *testing.T, count int, filename string, parquetSch
rowBuilder.Reset() rowBuilder.Reset()
// generate random data // generate random data
recordValue := NewRecordValueBuilder(). recordValue := NewRecordValueBuilder().
AddLongValue("ID", 1+int64(i)).
AddLongValue("CreatedAt", 2+2*int64(i)).
AddRecordValue("Person", NewRecordValueBuilder().
AddStringValue("zName", fmt.Sprintf("john_%d", i)).
AddStringListValue("emails",
SetLongValue("ID", 1+int64(i)).
SetLongValue("CreatedAt", 2+2*int64(i)).
SetRecordValue("Person", NewRecordValueBuilder().
SetStringValue("zName", fmt.Sprintf("john_%d", i)).
SetStringListValue("emails",
fmt.Sprintf("john_%d@a.com", i), fmt.Sprintf("john_%d@a.com", i),
fmt.Sprintf("john_%d@b.com", i), fmt.Sprintf("john_%d@b.com", i),
fmt.Sprintf("john_%d@c.com", i), fmt.Sprintf("john_%d@c.com", i),
fmt.Sprintf("john_%d@d.com", i), fmt.Sprintf("john_%d@d.com", i),
fmt.Sprintf("john_%d@e.com", i))). fmt.Sprintf("john_%d@e.com", i))).
AddStringValue("Company", fmt.Sprintf("company_%d", i)).Build()
SetStringValue("Company", fmt.Sprintf("company_%d", i)).Build()
AddRecordValue(rowBuilder, recordType, parquetLevels, recordValue) AddRecordValue(rowBuilder, recordType, parquetLevels, recordValue)
if count < 10 { if count < 10 {

Write Preview
Loading…
Cancel
Save