Skip to content

单操作说明

Jump to bottom
vanson edited this page Nov 28, 2023 · 4 revisions

单操作说明

部分版本的Server可能不止支持某些可选参数

Insert

// Insert a record by rowKey.
Insert(ctx context.Context, tableName string, rowKey []*table.Column, mutateColumns []*table.Column, opts ...ObkvOption) (int64, error)

param

  • ctx: 用于控制接口超时
  • tableName: 表名
  • rowKey: 主键
  • mutateColumns: 主键外的其他要写入的列
  • opts: 扩展项, 按需填写

return

  • int64: affected row,影响的行数
  • error: 错误信息,包括错误码和错误message

可选参数

  • ObOperationOption.WithFilter: 您可以通过检查filter中的条件来决定此行是否应该插入, 更多关于Filter的内容可以参考Filter章节
  • ObOperationOption.WithScanRange: 只有在Insert中生效的参数,server将检查同分区该range包含的行是否满足上面Filter条件, 来决定是否插入改行

demo

// CREATE TABLE test(c1 bigint, c2 bigint, PRIMARY KEY(c1)) PARTITION BY hash(c1) partitions 2;
func main() {
	const (
		configUrl    = "xxx"
		fullUserName = "user@tenant#cluster"
		passWord     = ""
		sysUserName  = "sysUser"
		sysPassWord  = ""
		tableName    = "test"
	)

	cfg := config.NewDefaultClientConfig()
	cli, err := client.NewClient(configUrl, fullUserName, passWord, sysUserName, sysPassWord, cfg)
	if err != nil {
		panic(err)
	}

	ctx, _ := context.WithTimeout(context.Background(), time.Duration(1000)*time.Millisecond) // 1000ms
	rowKey := []*table.Column{table.NewColumn("c1", int64(1))}
	mutateColumns := []*table.Column{table.NewColumn("c2", int64(2))}

	startRowKey := []*table.Column{table.NewColumn("c1", "1")}
	endRowKey := []*table.Column{table.NewColumn("c1", "10")}
	keyRanges := []*table.RangePair{table.NewRangePair(startRowKey, endRowKey)}

	affectRows, err := cli.Insert(
		ctx,
		tableName,
		rowKey,
		mutateColumns,
		option.WithFilter(filter.CompareVal(filter.Equal, "c2", int64(1))), // where c2 = 1
		option.WithScanRange(keyRanges),
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(affectRows)
}
  • 在demo中,我们往test表中写入了(1, 2), 其中1为主键c1的值，2为普通列c2的值
  • 在demo中,我们检查与(1, 2)同分区的范围为1 - 10的所有行是否有c2 = 1的行,有的话我们写入(1, 2)
  • 在demo中,我们通过context控制了Insert接口的执行时间最大为1000ms

Get

// Get a record by rowKey.
Get(ctx context.Context, tableName string, rowKey []*table.Column, getColumns []string, opts ...ObkvOption) (SingleResult, error)

param

  • ctx: 用于控制接口超时
  • tableName: 表名
  • rowKey: 主键
  • getColumns: 需要获取的列的列名, 填nil返回所有列
  • opts: 扩展项, 不需要填写

return

  • SingleResult: 获取的结果, 可以通过Value(columnName string) interface{}方法获取对应的列值结果
  • error: 错误信息, 包括错误码和错误message
type SingleResult interface {
	AffectedRows() int64
	Value(columnName string) interface{}
	RowKey() []interface{}
}
  • Value方法中的columnName大小写不敏感

demo

// CREATE TABLE test(c1 bigint, c2 bigint, PRIMARY KEY(c1)) PARTITION BY hash(c1) partitions 2;
func main() {
	// Init client
	// ...

	// insert
	ctx, _ := context.WithTimeout(context.Background(), time.Duration(1000)*time.Millisecond) // 1000ms
	rowKey := []*table.Column{table.NewColumn("c1", int64(1))}
	insertColumns := []*table.Column{table.NewColumn("c2", int64(2))}
	affectRows, err := cli.Insert(
		ctx,
		tableName,
		rowKey,
		insertColumns,
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(affectRows)

	// get
	selectColumns := []string{"c1", "c2"}
	result, err := cli.Get(
		ctx,
		tableName,
		rowKey,
		selectColumns,
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(result.Value("c1"))
	fmt.Println(result.Value("c2"))
}
  • 在demo中, 我们往test表中写入了(1, 2), 其中1为主键c1的值, 2为普通列c2的值
  • 在demo中, 我们通过rowkey(c1=1)获取了记录, 通过Value获取到'c1'和'c2'的值

Delete

// Delete a record by rowKey.
Delete(ctx context.Context, tableName string, rowKey []*table.Column, opts ...ObkvOption) (int64, error)

param

  • ctx: 用于控制接口超时
  • tableName: 表名
  • rowKey: 主键
  • opts: 扩展项，不需要填写

return

  • int64: affected row, 影响的列数
  • error: 错误信息, 包括错误码和错误message

可选参数

  • ObOperationOption.WithFilter: 您可以通过检查filter中的条件来决定此行是否应该删除, 更多关于Filter的内容可以参考Filter章节

demo

// CREATE TABLE test(c1 bigint, c2 bigint, PRIMARY KEY(c1)) PARTITION BY hash(c1) partitions 2;
func main() {
	// Init client
	// ...

	// insert
	ctx, _ := context.WithTimeout(context.Background(), time.Duration(1000)*time.Millisecond) // 1000ms
	rowKey := []*table.Column{table.NewColumn("c1", int64(1))}
	insertColumns := []*table.Column{table.NewColumn("c2", int64(2))}
	affectRows, err := cli.Insert(
		ctx,
		tableName,
		rowKey,
		insertColumns,
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(affectRows)

	// delete
	affectRows, err = cli.Delete(
		ctx,
		tableName,
		rowKey,
		option.WithFilter(filter.CompareVal(filter.Equal, "c2", int64(2))),
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(affectRows)
}
  • 在demo中, 我们往test表中写入了(1, 2), 其中1为主键c1的值，2为普通列c2的值
  • 在demo中, 我们删除rowkey(c1=1)且满足c2=2的记录

Update

// Update a record by rowKey.
Update(ctx context.Context, tableName string, rowKey []*table.Column, mutateColumns []*table.Column, opts ...ObkvOption) (int64, error)

param

  • ctx: 用于控制接口超时
  • tableName: 表名
  • rowKey: 主键
  • mutateColumns: 主键外的其他要更新的列
  • opts: 扩展项，不需要填写

return

  • int64: affected row, 影响的行数
  • error: 错误信息，包括错误码和错误message

可选参数

  • ObOperationOption.WithFilter: 您可以通过检查filter中的条件来决定此行是否应该插入, 更多关于Filter的内容可以参考Filter章节

demo

// CREATE TABLE test(c1 bigint, c2 bigint, PRIMARY KEY(c1)) PARTITION BY hash(c1) partitions 2;
func main() {
	// Init client
	// ...

	// insert
	ctx, _ := context.WithTimeout(context.Background(), time.Duration(1000)*time.Millisecond) // 1000ms
	rowKey := []*table.Column{table.NewColumn("c1", int64(1))}
	insertColumns := []*table.Column{table.NewColumn("c2", int64(2))}
	affectRows, err := cli.Insert(
		ctx,
		tableName,
		rowKey,
		insertColumns,
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(affectRows)

	// update c2(2) -> c2(3)
	updateColumns := []*table.Column{table.NewColumn("c2", int64(3))}
	affectRows, err = cli.Update(
		ctx,
		tableName,
		rowKey,
		updateColumns,
		option.WithFilter(filter.CompareVal(filter.Equal, "c2", int64(3))),
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(affectRows)
}
  • 在demo中, 我们往test表中写入了(1, 2), 其中1为主键c1的值，2为普通列c2的值
  • 在demo中, 我们更新了test表中rowkey(c1)为1的的记录, 如果c1=1这行的c2=3,那么我们将c2从2更新为3

InsertOrUpdate

// InsertOrUpdate insert or update a record by rowKey.
// insert if the primary key does not exist, update if it does.
InsertOrUpdate(ctx context.Context, tableName string, rowKey []*table.Column, mutateColumns []*table.Column, opts ...ObkvOption) (int64, error)

param

  • ctx: 用于控制接口超时
  • tableName: 表名
  • rowKey: 主键
  • mutateColumns: 主键外的其他要写入或更新的列
  • opts: 扩展项, 不需要填写

return

  • int64: affected row, 影响的行数
  • error: 错误信息, 包括错误码和错误message

注意事项

  • 由于InsertOrUpdate的语义问题，有可能是做insert操作有可能是做update操作，因此带有not null属性的列都需要填值，除非这个列还带有default value属性

demo

// CREATE TABLE test(c1 bigint, c2 bigint, PRIMARY KEY(c1)) PARTITION BY hash(c1) partitions 2;
func main() {
	// Init client
	// ...

	// insert
	ctx, _ := context.WithTimeout(context.Background(), time.Duration(1000)*time.Millisecond) // 1000ms
	rowKey := []*table.Column{table.NewColumn("c1", int64(1))}
	mutateColumns := []*table.Column{table.NewColumn("c2", int64(2))}
	affectRows, err := cli.InsertOrUpdate(
		ctx,
		tableName,
		rowKey,
		mutateColumns,
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(affectRows)

	// update
	mutateColumns = []*table.Column{table.NewColumn("c2", int64(3))}
	affectRows, err = cli.InsertOrUpdate(
		ctx,
		tableName,
		rowKey,
		mutateColumns,
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(affectRows)
}
  • 在demo中, 我们调用InsertOrUpdate接口往test表中写入了(1, 2), 其中1为主键c1的值, 2为普通列c2的值
  • 在demo中, 我们调用InsertOrUpdate接口更新了test表中rowkey(c1)为1的的记录, 将c2从2更新为3

Replace

// Replace a record by rowKey.
Replace(ctx context.Context, tableName string, rowKey []*table.Column, mutateColumns []*table.Column, opts ...ObkvOption) (int64, error)

param

  • ctx: 用于控制接口超时
  • tableName: 表名
  • rowKey: 主键
  • mutateColumns: 主键外的其他要替换的列
  • opts: 扩展项, 不需要填写

return

  • int64: affected row, 影响的行数
  • error: 错误信息, 包括错误码和错误message

demo

// CREATE TABLE test(c1 bigint, c2 bigint, PRIMARY KEY(c1)) PARTITION BY hash(c1) partitions 2;
func main() {
	// Init client
	// ...

	// insert
	ctx, _ := context.WithTimeout(context.Background(), time.Duration(1000)*time.Millisecond) // 1000ms
	rowKey := []*table.Column{table.NewColumn("c1", int64(1))}
	insertColumns := []*table.Column{table.NewColumn("c2", int64(2))}
	affectRows, err := cli.Insert(
		ctx,
		tableName,
		rowKey,
		insertColumns,
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(affectRows)

	// replace c2(2) -> c2(3)
	replaceColumns := []*table.Column{table.NewColumn("c2", int64(3))}
	affectRows, err = cli.Replace(
		ctx,
		tableName,
		rowKey,
		replaceColumns,
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(affectRows)
}
  • 在demo中, 我们往test表中写入了(1, 2), 其中1为主键c1的值, 2为普通列c2的值
  • 在demo中, 我们替换了test表中rowkey(c1)为1的的记录, 将c2从2替换为3

Increment

// Increment a record by rowKey.
Increment(ctx context.Context, tableName string, rowKey []*table.Column, mutateColumns []*table.Column, opts ...ObkvOption) (SingleResult, error)

param

  • ctx: 用于控制接口超时
  • tableName: 表名
  • rowKey: 主键
  • mutateColumns: 主键外的其他要increment的列
  • opts: 扩展项, 按需填写

return

  • int64: affected row, 影响的行数
  • error: 错误信息, 包括错误码和错误message

可选参数

  • ObOperationOption.WithReturnRowKey: 是否需要返回主键
  • ObOperationOption.WithReturnAffectedEntity: 是否需要返回被改变的原始行
  • ObOperationOption.WithFilter: 您可以通过检查filter中的条件来决定此行是否应该插入, 更多关于Filter的内容可以参考Filter章节

constraint

  • 只支持在signed列上进行increment

demo

// CREATE TABLE test(c1 bigint, c2 bigint, PRIMARY KEY(c1)) PARTITION BY hash(c1) partitions 2;
func main() {
	// Init client
	// ...

	// insert
	ctx, _ := context.WithTimeout(context.Background(), time.Duration(1000)*time.Millisecond) // 1000ms
	rowKey := []*table.Column{table.NewColumn("c1", int64(1))}
	insertColumns := []*table.Column{table.NewColumn("c2", int64(2))}
	affectRows, err := cli.Insert(
		ctx,
		tableName,
		rowKey,
		insertColumns,
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(affectRows)

	// increment c2(2) -> c2(2) + (3) = c2(5)
	incrementColumns := []*table.Column{table.NewColumn("c2", int64(3))}
	res, err := cli.Increment(
		ctx,
		tableName,
		rowKey,
		incrementColumns,
		client.WithReturnRowKey(true),         // return rowKey if you need
		client.WithReturnAffectedEntity(true), // return result entity if you need
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(res.AffectedRows())
	fmt.Println(res.RowKey())
	fmt.Println(res.Value("c2"))
}
  • 在demo中，我们往test表中写入了(1, 2), 其中1为主键c1的值，2为普通列c2的值
  • 在demo中，我们通过主键c1=1, 将c2的值增加3
  • 在demo中，我们通过WithReturnRowKey(true)返回了主键
  • 在demo中，我们通过WithReturnAffectedEntity(true)返回了修改后的列, 知道c2被修改成了5

Append

// Append a record by rowKey.
Append(ctx context.Context, tableName string, rowKey []*table.Column, mutateColumns []*table.Column, opts ...ObkvOption) (SingleResult, error)

param

  • ctx: 用于控制接口超时
  • tableName: 表名
  • rowKey: 主键
  • mutateColumns: 主键外的其他的append的列
  • opts: 扩展项, 按需填写

return

  • int64: affected row, 影响的行数
  • error: 错误信息, 包括错误码和错误message

可选参数

  • ObOperationOption.WithReturnRowKey: 是否需要返回主键
  • ObOperationOption.WithReturnAffectedEntity: 是否需要返回被改变的原始行
  • ObOperationOption.WithFilter: 您可以通过检查filter中的条件来决定此行是否应该插入, 更多关于Filter的内容可以参考Filter章节

constraint

  • observer 3.x只支持在varchar列上进行append
  • observer 4.x版本支持在char、varchar列上进行append, append后列长度不允许超过1M

demo

// CREATE TABLE test(c1 bigint, c2 varchar(20), PRIMARY KEY(c1)) PARTITION BY hash(c1) partitions 2;
func main() {
	// Init client
	// ...

	// insert
	ctx, _ := context.WithTimeout(context.Background(), time.Duration(1000)*time.Millisecond) // 1000ms
	rowKey := []*table.Column{table.NewColumn("c1", int64(1))}
	insertColumns := []*table.Column{table.NewColumn("c2", "hello")}
	affectRows, err := cli.Insert(
		ctx,
		tableName,
		rowKey,
		insertColumns,
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(affectRows)

	// append c2(2) -> c2("hello") + (" oceanbase") = c2("hello oceanbase")
	appendColumns := []*table.Column{table.NewColumn("c2", " oceanbase")}
	res, err := cli.Append(
		ctx,
		tableName,
		rowKey,
		appendColumns,
		client.WithReturnRowKey(true),         // return rowKey if you need
		client.WithReturnAffectedEntity(true), // return result entity if you need
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(res.AffectedRows())
	fmt.Println(res.RowKey())
	fmt.Println(res.Value("c2"))
}
  • 在demo中, 我们往test表中写入了(1, 'hello'), 其中1为主键c1的值, 'hello'为普通列c2的值
  • 在demo中, 我们通过主键c1=1, 将c2的值追加' oceanbase'
  • 在demo中, 我们通过WithReturnRowKey(true)返回了主键
  • 在demo中, 我们通过WithReturnAffectedEntity(true)返回了修改后的列, 知道c2被修改成了'hello oceanbase'
Clone this wiki locally