Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add AppendRow function to BatchColumn #927

Merged
merged 5 commits into from
Mar 8, 2023
Merged

Add AppendRow function to BatchColumn #927

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ac5b9a1
AppendRow is visible for columnar batch requests
pikot Mar 2, 2023
ff20e5a
fix columnar_batch_test fro AppendRow, and parseDate due to error in …
pikot Mar 2, 2023
aed0590
fix tests
pikot Mar 6, 2023
82b0ebb
fix string formation
pikot Mar 6, 2023
413567e
merge tz to date
pikot Mar 6, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions conn_batch.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,6 +225,21 @@ func (b *batchColumn) Append(v interface{}) (err error) {
return nil
}

func (b *batchColumn) AppendRow(v interface{}) (err error) {
if b.batch.IsSent() {
return ErrBatchAlreadySent
}
if b.err != nil {
b.release(b.err)
return b.err
}
if err = b.column.AppendRow(v); err != nil {
b.release(err)
return err
}
return nil
}

var (
_ (driver.Batch) = (*batch)(nil)
_ (driver.BatchColumn) = (*batchColumn)(nil)
Expand Down
1 change: 1 addition & 0 deletions lib/driver/driver.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@ type (
}
BatchColumn interface {
Append(interface{}) error
AppendRow(interface{}) error
}
ColumnType interface {
Name() string
Expand Down
198 changes: 198 additions & 0 deletions tests/columnar_batch_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -250,3 +250,201 @@ func TestNullableColumnarInterface(t *testing.T) {
assert.Equal(t, uint64(20), count)
}
}

func TestColumnarAppendRowInterface(t *testing.T) {
conn, err := GetNativeConnection(nil, nil, &clickhouse.Compression{
Method: clickhouse.CompressionLZ4,
})
ctx := context.Background()
require.NoError(t, err)
const ddl = `
CREATE TABLE test_column_interface (
Col1 UInt8
, Col2 String
, Col3 DateTime
, Col4 String
, Col5 DateTime
, Col6 Int64
) Engine MergeTree() ORDER BY tuple()
`
defer func() {
conn.Exec(ctx, "DROP TABLE IF EXISTS test_column_interface")
}()
require.NoError(t, conn.Exec(ctx, ddl))
batch, err := conn.PrepareBatch(ctx, "INSERT INTO test_column_interface")
require.NoError(t, err)
var currentTime = time.Now().Truncate(time.Second)

for i := 0; i < 150; i++ {
require.NoError(t, batch.Column(0).AppendRow(uint8(i)))
require.NoError(t, batch.Column(1).AppendRow(fmt.Sprintf("value_%d", i)))
require.NoError(t, batch.Column(2).AppendRow(currentTime))
require.NoError(t, batch.Column(3).AppendRow(sql.NullString{String: fmt.Sprintf("value_%d", i), Valid: true}))
require.NoError(t, batch.Column(4).AppendRow(sql.NullTime{Time: currentTime, Valid: true}))
require.NoError(t, batch.Column(5).AppendRow( sql.NullInt64{Int64: int64(i), Valid: true}))
}
require.NoError(t, batch.Send())
var count uint64
require.NoError(t, conn.QueryRow(ctx, "SELECT COUNT() FROM test_column_interface").Scan(&count))
require.Equal(t, uint64(150), count)
rows, err := conn.Query(ctx, "SELECT * FROM test_column_interface WHERE Col1 >= $1 AND Col1 < $2", 10, 30)
require.NoError(t, err)
var (
row uint8 = 10
)
iCount := 0
for rows.Next() {
var (
col1 uint8
col2 string
col3 time.Time
col4 sql.NullString
col5 sql.NullTime
col6 sql.NullInt64
)
require.NoError(t, rows.Scan(&col1, &col2, &col3, &col4, &col5, &col6))
assert.Equal(t, row, col1)
assert.Equal(t, fmt.Sprintf("value_%d", row), col2)
assert.Equal(t, currentTime.Unix(), col3.Unix())
assert.Equal(t, fmt.Sprintf("value_%d", row), col4.String)
assert.Equal(t, currentTime.In(time.UTC), col5.Time)
assert.Equal(t, int64(row), col6.Int64)
row++
iCount++
}
rows.Close()
require.NoError(t, rows.Err())
assert.Equal(t, 20, iCount)
}

func TestNullableAppendRowColumnarInterface(t *testing.T) {
conn, err := GetNativeConnection(nil, nil, &clickhouse.Compression{
Method: clickhouse.CompressionLZ4,
})
ctx := context.Background()
require.NoError(t, err)
const ddl = `
CREATE TABLE test_column_interface (
Col1 Nullable(UInt8)
, Col2 Nullable(String)
, Col3 Nullable(DateTime)
, Col4 Nullable(Decimal(10, 2))
) Engine MergeTree() ORDER BY tuple()
`
defer func() {
conn.Exec(ctx, "DROP TABLE IF EXISTS test_column_interface")
}()
require.NoError(t, conn.Exec(ctx, ddl))
batch, err := conn.PrepareBatch(ctx, "INSERT INTO test_column_interface")
require.NoError(t, err)

var (
currentTime = time.Now().Truncate(time.Second)
decimalVal = decimal.NewFromFloat(12.02)
)

for i := 0; i < 150; i++ {
a, b := uint8(i), fmt.Sprintf("value_%d", i)
{
require.NoError(t, batch.Column(0).AppendRow(&a))
require.NoError(t, batch.Column(1).AppendRow(&b))
require.NoError(t, batch.Column(2).AppendRow(&currentTime))
require.NoError(t, batch.Column(3).AppendRow(&decimalVal))
}
}
require.NoError(t, batch.Send())
var count uint64
require.NoError(t, conn.QueryRow(ctx, "SELECT COUNT() FROM test_column_interface").Scan(&count))
require.Equal(t, uint64(150), count)
rows, err := conn.Query(ctx, "SELECT * FROM test_column_interface WHERE Col1 >= $1 AND Col1 < $2", 10, 30)
require.NoError(t, err)
var (
row uint8 = 10
)
count = 0
for rows.Next() {
var (
col1 *uint8
col2 *string
col3 *time.Time
col4 *decimal.Decimal
)
if assert.NoError(t, rows.Scan(&col1, &col2, &col3, &col4)) {
assert.Equal(t, row, *col1)
assert.Equal(t, fmt.Sprintf("value_%d", row), *col2)
assert.Equal(t, currentTime.Unix(), col3.Unix())
assert.Equal(t, decimalVal.String(), (*col4).String())
}
row++
count++
}
rows.Close()
require.NoError(t, rows.Err())
assert.Equal(t, uint64(20), count)
require.NoError(t, conn.Exec(ctx, "TRUNCATE TABLE test_column_interface"))
batch, err = conn.PrepareBatch(ctx, "INSERT INTO test_column_interface")
require.NoError(t, err)
{

currentTime = time.Now().Truncate(time.Second)

for i := 0; i < 150; i++ {
a, b := uint8(i), fmt.Sprintf("value_%d", i)
require.NoError(t, batch.Column(0).AppendRow( &a ))

switch {
case i%2 == 0:
require.NoError(t, batch.Column(1).AppendRow(&b))
require.NoError(t, batch.Column(2).AppendRow(&currentTime))
require.NoError(t, batch.Column(3).AppendRow(&decimalVal))
default:
require.NoError(t, batch.Column(1).AppendRow(nil))
require.NoError(t, batch.Column(2).AppendRow(nil))
require.NoError(t, batch.Column(3).AppendRow(nil))
}
}
require.NoError(t, batch.Send())
var count uint64
require.NoError(t, conn.QueryRow(ctx, "SELECT COUNT() FROM test_column_interface").Scan(&count))
require.Equal(t, uint64(150), count)
rows, err := conn.Query(ctx, "SELECT * FROM test_column_interface WHERE Col1 >= $1 AND Col1 < $2", 10, 30)
require.NoError(t, err)
var (
row uint8 = 10
)
count = 0
for rows.Next() {
var (
col1 *uint8
col2 *string
col3 *time.Time
col4 *decimal.Decimal
)
require.NoError(t, rows.Scan(&col1, &col2, &col3, &col4))
switch {
case row%2 == 0:
assert.Equal(t, row, *col1)
assert.Equal(t, fmt.Sprintf("value_%d", row), *col2)
assert.Equal(t, currentTime.Unix(), col3.Unix())
assert.Equal(t, decimalVal.String(), (*col4).String())
default:
if assert.Equal(t, row, *col1) {
assert.Nil(t, col2)
assert.Nil(t, col3)
assert.Nil(t, col4)
}
}
row++
count++
}
rows.Close()
require.NoError(t, rows.Err())
assert.Equal(t, uint64(20), count)
}
}






2 changes: 1 addition & 1 deletion tests/date32_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,7 +157,7 @@ func TestNullableDate32(t *testing.T) {
date, err := time.Parse("2006-01-02 15:04:05", "2283-11-11 00:00:00")
require.NoError(t, err)
dateStr := "2283-11-11"
require.NoError(t, batch.Append(date, date, dateStr, &dateStr))
require.NoError(t, batch.Append(date, &date, dateStr, &dateStr))
require.NoError(t, batch.Send())
var (
col1 *time.Time
Expand Down