|
6 | 6 | "path" |
7 | 7 | "time" |
8 | 8 |
|
| 9 | + "google.golang.org/grpc" |
| 10 | + "google.golang.org/grpc/encoding/gzip" |
| 11 | + |
9 | 12 | "github.com/ydb-platform/ydb-go-sdk/v3" |
10 | 13 | "github.com/ydb-platform/ydb-go-sdk/v3/retry" |
11 | 14 | "github.com/ydb-platform/ydb-go-sdk/v3/table" |
@@ -110,18 +113,13 @@ func Example_bulkUpsert() { |
110 | 113 | const batchSize = 10000 |
111 | 114 | logs := make([]logMessage, 0, batchSize) |
112 | 115 | for i := 0; i < batchSize; i++ { |
113 | | - message := logMessage{ |
| 116 | + logs = append(logs, logMessage{ |
114 | 117 | App: fmt.Sprintf("App_%d", i/256), |
115 | 118 | Host: fmt.Sprintf("192.168.0.%d", i%256), |
116 | 119 | Timestamp: time.Now().Add(time.Millisecond * time.Duration(i%1000)), |
117 | 120 | HTTPCode: 200, |
118 | | - } |
119 | | - if i%2 == 0 { |
120 | | - message.Message = "GET / HTTP/1.1" |
121 | | - } else { |
122 | | - message.Message = "GET /images/logo.png HTTP/1.1" |
123 | | - } |
124 | | - logs = append(logs, message) |
| 121 | + Message: "GET / HTTP/1.1", |
| 122 | + }) |
125 | 123 | } |
126 | 124 | // execute bulk upsert with native ydb data |
127 | 125 | err = db.Table().Do( // Do retry operation on errors with best effort |
@@ -250,3 +248,149 @@ func Example_lazyTransaction() { |
250 | 248 | fmt.Printf("unexpected error: %v", err) |
251 | 249 | } |
252 | 250 | } |
| 251 | + |
| 252 | +func Example_bulkUpsertWithCompression() { |
| 253 | + ctx := context.TODO() |
| 254 | + db, err := ydb.Open(ctx, "grpc://localhost:2136/local") |
| 255 | + if err != nil { |
| 256 | + fmt.Printf("failed connect: %v", err) |
| 257 | + return |
| 258 | + } |
| 259 | + defer db.Close(ctx) // cleanup resources |
| 260 | + type logMessage struct { |
| 261 | + App string |
| 262 | + Host string |
| 263 | + Timestamp time.Time |
| 264 | + HTTPCode uint32 |
| 265 | + Message string |
| 266 | + } |
| 267 | + // prepare native go data |
| 268 | + const batchSize = 10000 |
| 269 | + logs := make([]logMessage, 0, batchSize) |
| 270 | + for i := 0; i < batchSize; i++ { |
| 271 | + logs = append(logs, logMessage{ |
| 272 | + App: fmt.Sprintf("App_%d", i/256), |
| 273 | + Host: fmt.Sprintf("192.168.0.%d", i%256), |
| 274 | + Timestamp: time.Now().Add(time.Millisecond * time.Duration(i%1000)), |
| 275 | + HTTPCode: 200, |
| 276 | + Message: "GET /images/logo.png HTTP/1.1", |
| 277 | + }) |
| 278 | + } |
| 279 | + // execute bulk upsert with native ydb data |
| 280 | + err = db.Table().Do( // Do retry operation on errors with best effort |
| 281 | + ctx, // context manage exiting from Do |
| 282 | + func(ctx context.Context, s table.Session) (err error) { // retry operation |
| 283 | + rows := make([]types.Value, 0, len(logs)) |
| 284 | + for _, msg := range logs { |
| 285 | + rows = append(rows, types.StructValue( |
| 286 | + types.StructFieldValue("App", types.TextValue(msg.App)), |
| 287 | + types.StructFieldValue("Host", types.TextValue(msg.Host)), |
| 288 | + types.StructFieldValue("Timestamp", types.TimestampValueFromTime(msg.Timestamp)), |
| 289 | + types.StructFieldValue("HTTPCode", types.Uint32Value(msg.HTTPCode)), |
| 290 | + types.StructFieldValue("Message", types.TextValue(msg.Message)), |
| 291 | + )) |
| 292 | + } |
| 293 | + return s.BulkUpsert(ctx, "/local/bulk_upsert_example", types.ListValue(rows...), |
| 294 | + options.WithCallOptions(grpc.UseCompressor(gzip.Name)), |
| 295 | + ) |
| 296 | + }, |
| 297 | + table.WithIdempotent(), |
| 298 | + ) |
| 299 | + if err != nil { |
| 300 | + fmt.Printf("unexpected error: %v", err) |
| 301 | + } |
| 302 | +} |
| 303 | + |
| 304 | +func Example_dataQueryWithCompression() { |
| 305 | + ctx := context.TODO() |
| 306 | + db, err := ydb.Open(ctx, "grpc://localhost:2136/local") |
| 307 | + if err != nil { |
| 308 | + fmt.Printf("failed connect: %v", err) |
| 309 | + return |
| 310 | + } |
| 311 | + defer db.Close(ctx) // cleanup resources |
| 312 | + var ( |
| 313 | + query = `SELECT 42 as id, "my string" as myStr` |
| 314 | + id int32 // required value |
| 315 | + myStr string // optional value |
| 316 | + ) |
| 317 | + err = db.Table().Do( // Do retry operation on errors with best effort |
| 318 | + ctx, // context manage exiting from Do |
| 319 | + func(ctx context.Context, s table.Session) (err error) { // retry operation |
| 320 | + _, res, err := s.Execute(ctx, table.DefaultTxControl(), query, nil, |
| 321 | + options.WithCallOptions( |
| 322 | + grpc.UseCompressor(gzip.Name), |
| 323 | + ), |
| 324 | + ) |
| 325 | + if err != nil { |
| 326 | + return err // for auto-retry with driver |
| 327 | + } |
| 328 | + defer res.Close() // cleanup resources |
| 329 | + if err = res.NextResultSetErr(ctx); err != nil { // check single result set and switch to it |
| 330 | + return err // for auto-retry with driver |
| 331 | + } |
| 332 | + for res.NextRow() { // iterate over rows |
| 333 | + err = res.ScanNamed( |
| 334 | + named.Required("id", &id), |
| 335 | + named.OptionalWithDefault("myStr", &myStr), |
| 336 | + ) |
| 337 | + if err != nil { |
| 338 | + return err // generally scan error not retryable, return it for driver check error |
| 339 | + } |
| 340 | + fmt.Printf("id=%v, myStr='%s'\n", id, myStr) |
| 341 | + } |
| 342 | + return res.Err() // return finally result error for auto-retry with driver |
| 343 | + }, |
| 344 | + table.WithIdempotent(), |
| 345 | + ) |
| 346 | + if err != nil { |
| 347 | + fmt.Printf("unexpected error: %v", err) |
| 348 | + } |
| 349 | +} |
| 350 | + |
| 351 | +func Example_scanQueryWithCompression() { |
| 352 | + ctx := context.TODO() |
| 353 | + db, err := ydb.Open(ctx, "grpc://localhost:2136/local") |
| 354 | + if err != nil { |
| 355 | + fmt.Printf("failed connect: %v", err) |
| 356 | + return |
| 357 | + } |
| 358 | + defer db.Close(ctx) // cleanup resources |
| 359 | + var ( |
| 360 | + query = `SELECT 42 as id, "my string" as myStr` |
| 361 | + id int32 // required value |
| 362 | + myStr string // optional value |
| 363 | + ) |
| 364 | + err = db.Table().Do( // Do retry operation on errors with best effort |
| 365 | + ctx, // context manage exiting from Do |
| 366 | + func(ctx context.Context, s table.Session) (err error) { // retry operation |
| 367 | + res, err := s.StreamExecuteScanQuery(ctx, query, nil, |
| 368 | + options.WithCallOptions( |
| 369 | + grpc.UseCompressor(gzip.Name), |
| 370 | + ), |
| 371 | + ) |
| 372 | + if err != nil { |
| 373 | + return err // for auto-retry with driver |
| 374 | + } |
| 375 | + defer res.Close() // cleanup resources |
| 376 | + if err = res.NextResultSetErr(ctx); err != nil { // check single result set and switch to it |
| 377 | + return err // for auto-retry with driver |
| 378 | + } |
| 379 | + for res.NextRow() { // iterate over rows |
| 380 | + err = res.ScanNamed( |
| 381 | + named.Required("id", &id), |
| 382 | + named.OptionalWithDefault("myStr", &myStr), |
| 383 | + ) |
| 384 | + if err != nil { |
| 385 | + return err // generally scan error not retryable, return it for driver check error |
| 386 | + } |
| 387 | + fmt.Printf("id=%v, myStr='%s'\n", id, myStr) |
| 388 | + } |
| 389 | + return res.Err() // return finally result error for auto-retry with driver |
| 390 | + }, |
| 391 | + table.WithIdempotent(), |
| 392 | + ) |
| 393 | + if err != nil { |
| 394 | + fmt.Printf("unexpected error: %v", err) |
| 395 | + } |
| 396 | +} |
0 commit comments