Add idempotent flag to all calls of db.Table().Do()

asmyasnikov · asmyasnikov · commit b053d3144a6e · 2022-11-18T15:00:18.000+03:00
diff --git a/example_test.go b/example_test.go
@@ -49,6 +49,7 @@ func Example_table() {
 			}
 			return res.Err() // return finally result error for auto-retry with driver
 		},
+		table.WithIdempotent(),
 	)
 	if err != nil {
 		log.Printf("unexpected error: %v", err)
diff --git a/meta/example_test.go b/meta/example_test.go
@@ -50,6 +50,7 @@ func Example_consumedUnitsCount() {
 			}
 			return res.Err() // return finally result error for auto-retry with driver
 		},
+		table.WithIdempotent(),
 	)
 	if err != nil {
 		log.Printf("unexpected error: %v", err)
diff --git a/sugar/path.go b/sugar/path.go
@@ -115,7 +115,7 @@ func RemoveRecursive(ctx context.Context, db ydb.Connection, pathToRemove string
 			case scheme.EntryTable:
 				err = db.Table().Do(ctx, func(ctx context.Context, session table.Session) (err error) {
 					return session.DropTable(ctx, pt)
-				})
+				}, table.WithIdempotent())
 				if err != nil {
 					return xerrors.WithStackTrace(err)
 				}
diff --git a/table/example_test.go b/table/example_test.go
@@ -49,6 +49,7 @@ func Example_select() {
 			}
 			return res.Err() // return finally result error for auto-retry with driver
 		},
+		table.WithIdempotent(),
 	)
 	if err != nil {
 		fmt.Printf("unexpected error: %v", err)
@@ -78,6 +79,7 @@ func Example_createTable() {
 				),
 			)
 		},
+		table.WithIdempotent(),
 	)
 	if err != nil {
 		fmt.Printf("unexpected error: %v", err)
@@ -132,6 +134,7 @@ func Example_bulkUpsert() {
 			}
 			return s.BulkUpsert(ctx, "/local/bulk_upsert_example", types.ListValue(rows...))
 		},
+		table.WithIdempotent(),
 	)
 	if err != nil {
 		fmt.Printf("unexpected error: %v", err)
@@ -162,6 +165,7 @@ func Example_alterTable() {
 				options.WithDropAttribute("baz"),
 			)
 		},
+		table.WithIdempotent(),
 	)
 	if err != nil {
 		fmt.Printf("unexpected error: %v", err)
diff --git a/table/table_e2e_test.go b/table/table_e2e_test.go
@@ -1264,6 +1264,7 @@ func fill(ctx context.Context, db ydb.Connection, folder string) error {
 			))
 			return
 		},
+		table.WithIdempotent(),
 	)
 }
 
@@ -1302,6 +1303,7 @@ func createTables(ctx context.Context, c table.Client, folder string) error {
 				options.WithPrimaryKeyColumn("series_id", "season_id"),
 			)
 		},
+		table.WithIdempotent(),
 	)
 	if err != nil {
 		return err
@@ -1322,6 +1324,7 @@ func createTables(ctx context.Context, c table.Client, folder string) error {
 				options.WithPrimaryKeyColumn("series_id", "season_id", "episode_id"),
 			)
 		},
+		table.WithIdempotent(),
 	)
 	return err
 }
@@ -1399,6 +1402,7 @@ func TestLongStream(t *testing.T) {
 					`CREATE TABLE `+tableName+` (val Int64, PRIMARY KEY (val))`,
 				)
 			},
+			table.WithIdempotent(),
 		); err != nil {
 			t.Fatalf("create table failed: %v\n", err)
 		}
@@ -1454,6 +1458,7 @@ func TestLongStream(t *testing.T) {
 						)
 						return err
 					},
+					table.WithIdempotent(),
 				); err != nil {
 					t.Fatalf("upsert failed: %v\n", err)
 				} else {
@@ -1517,6 +1522,7 @@ func TestLongStream(t *testing.T) {
 				}
 				return nil
 			},
+			table.WithIdempotent(),
 		); err != nil {
 			t.Fatalf("stream query failed: %v\n", err)
 		}
@@ -1557,6 +1563,7 @@ func TestLongStream(t *testing.T) {
 				}
 				return nil
 			},
+			table.WithIdempotent(),
 		); err != nil {
 			t.Fatalf("stream query failed: %v\n", err)
 		}
@@ -1611,6 +1618,7 @@ func TestSplitRangesAndRead(t *testing.T) {
 					)`,
 				)
 			},
+			table.WithIdempotent(),
 		); err != nil {
 			t.Fatalf("create table failed: %v\n", err)
 		}
@@ -1666,6 +1674,7 @@ func TestSplitRangesAndRead(t *testing.T) {
 						)
 						return err
 					},
+					table.WithIdempotent(),
 				); err != nil {
 					t.Fatalf("upsert failed: %v\n", err)
 				} else {
@@ -1724,6 +1733,7 @@ func TestSplitRangesAndRead(t *testing.T) {
 				}
 				return nil
 			},
+			table.WithIdempotent(),
 		); err != nil {
 			t.Fatalf("stream query failed: %v\n", err)
 		}
@@ -1757,6 +1767,7 @@ func TestSplitRangesAndRead(t *testing.T) {
 					}
 					return nil
 				},
+				table.WithIdempotent(),
 			); err != nil {
 				t.Fatalf("stream query failed: %v\n", err)
 			}
@@ -1788,14 +1799,18 @@ func TestIssue229UnexpectedNullWhileParseNilJsonDocumentValue(t *testing.T) {
 	defer db.Close(ctx)
 	err := db.Table().DoTx(ctx, func(ctx context.Context, tx table.TransactionActor) error {
 		res, err := tx.Execute(ctx, `SELECT Nothing(JsonDocument?) AS r`, nil)
-		require.NoError(t, err)
-		require.NoError(t, res.NextResultSetErr(ctx))
+		if err != nil {
+			return err
+		}
+		if err = res.NextResultSetErr(ctx); err != nil {
+			return err
+		}
 		require.True(t, res.NextRow())
 
 		var val issue229Struct
 		require.NoError(t, res.Scan(&val))
-		return nil
-	})
+		return res.Err()
+	}, table.WithIdempotent())
 	require.NoError(t, err)
 }
 
@@ -1822,8 +1837,12 @@ func TestIssue259IntervalFromDuration(t *testing.T) {
 			SELECT $ts == $ten_micro, $ten_micro;`, table.NewQueryParameters(
 			table.ValueParam(`$ts`, types.IntervalValueFromDuration(10*time.Microsecond)),
 		))
-		require.NoError(t, err)
-		require.NoError(t, res.NextResultSetErr(ctx))
+		if err != nil {
+			return err
+		}
+		if err = res.NextResultSetErr(ctx); err != nil {
+			return err
+		}
 		require.True(t, res.NextRow())
 
 		var (
@@ -1950,7 +1969,7 @@ func TestNullType(t *testing.T) {
 		}
 		fmt.Printf("%+v\n", rescheduleDue)
 		return res.Err()
-	}, table.WithTxSettings(table.TxSettings(table.WithSnapshotReadOnly())))
+	}, table.WithTxSettings(table.TxSettings(table.WithSnapshotReadOnly())), table.WithIdempotent())
 	require.NoError(t, err)
 }
 
@@ -2193,7 +2212,7 @@ func TestValueToYqlLiteral(t *testing.T) {
 				require.Equal(t, 1, len(values))
 				require.Equal(t, tt.Yql(), values[0].Yql(), fmt.Sprintf("%T vs %T", tt, values[0]))
 				return nil
-			})
+			}, table.WithIdempotent())
 			require.NoError(t, err)
 		})
 	}
diff --git a/topic/cdc_reader_e2e_test.go b/topic/cdc_reader_e2e_test.go
@@ -226,10 +226,12 @@ func TestCDCInTableDescribe(t *testing.T) {
 			tablePath := path.Dir(topicPath)
 			topicName := path.Base(topicPath)
 			desc, err := s.DescribeTable(ctx, tablePath)
-			require.NoError(t, err)
+			if err != nil {
+				return err
+			}
 			require.Equal(t, topicName, desc.Changefeeds[0].Name)
 			return nil
-		})
+		}, table.WithIdempotent())
 		require.NoError(t, err)
 	})
 }
@@ -265,7 +267,7 @@ func createCDCFeed(ctx context.Context, t *testing.T, db ydb.Connection) string
 		}
 
 		return nil
-	})
+	}, table.WithIdempotent())
 	require.NoError(t, err)
 
 	topicPath := testCDCFeedName(db)

Original file line number	Diff line number	Diff line change
`@@ -49,6 +49,7 @@ func Example_table() {`
`49`	`49`	`}`
`50`	`50`	`return res.Err() // return finally result error for auto-retry with driver`
`51`	`51`	`},`
	`52`	`+ table.WithIdempotent(),`
`52`	`53`	`)`
`53`	`54`	`if err != nil {`
`54`	`55`	`log.Printf("unexpected error: %v", err)`
Original file line number	Diff line number	Diff line change
`@@ -50,6 +50,7 @@ func Example_consumedUnitsCount() {`
`50`	`50`	`}`
`51`	`51`	`return res.Err() // return finally result error for auto-retry with driver`
`52`	`52`	`},`
	`53`	`+ table.WithIdempotent(),`
`53`	`54`	`)`
`54`	`55`	`if err != nil {`
`55`	`56`	`log.Printf("unexpected error: %v", err)`
Original file line number	Diff line number	Diff line change
`@@ -115,7 +115,7 @@ func RemoveRecursive(ctx context.Context, db ydb.Connection, pathToRemove string`
`115`	`115`	`case scheme.EntryTable:`
`116`	`116`	`err = db.Table().Do(ctx, func(ctx context.Context, session table.Session) (err error) {`
`117`	`117`	`return session.DropTable(ctx, pt)`
`118`		`- })`
	`118`	`+ }, table.WithIdempotent())`
`119`	`119`	`if err != nil {`
`120`	`120`	`return xerrors.WithStackTrace(err)`
`121`	`121`	`}`
Original file line number	Diff line number	Diff line change
`@@ -49,6 +49,7 @@ func Example_select() {`
`49`	`49`	`}`
`50`	`50`	`return res.Err() // return finally result error for auto-retry with driver`
`51`	`51`	`},`
	`52`	`+ table.WithIdempotent(),`
`52`	`53`	`)`
`53`	`54`	`if err != nil {`
`54`	`55`	`fmt.Printf("unexpected error: %v", err)`
`@@ -78,6 +79,7 @@ func Example_createTable() {`
`78`	`79`	`),`
`79`	`80`	`)`
`80`	`81`	`},`
	`82`	`+ table.WithIdempotent(),`
`81`	`83`	`)`
`82`	`84`	`if err != nil {`
`83`	`85`	`fmt.Printf("unexpected error: %v", err)`
`@@ -132,6 +134,7 @@ func Example_bulkUpsert() {`
`132`	`134`	`}`
`133`	`135`	`return s.BulkUpsert(ctx, "/local/bulk_upsert_example", types.ListValue(rows...))`
`134`	`136`	`},`
	`137`	`+ table.WithIdempotent(),`
`135`	`138`	`)`
`136`	`139`	`if err != nil {`
`137`	`140`	`fmt.Printf("unexpected error: %v", err)`
`@@ -162,6 +165,7 @@ func Example_alterTable() {`
`162`	`165`	`options.WithDropAttribute("baz"),`
`163`	`166`	`)`
`164`	`167`	`},`
	`168`	`+ table.WithIdempotent(),`
`165`	`169`	`)`
`166`	`170`	`if err != nil {`
`167`	`171`	`fmt.Printf("unexpected error: %v", err)`
Original file line number	Diff line number	Diff line change
`@@ -1264,6 +1264,7 @@ func fill(ctx context.Context, db ydb.Connection, folder string) error {`
`1264`	`1264`	`))`
`1265`	`1265`	`return`
`1266`	`1266`	`},`
	`1267`	`+ table.WithIdempotent(),`
`1267`	`1268`	`)`
`1268`	`1269`	`}`
`1269`	`1270`
`@@ -1302,6 +1303,7 @@ func createTables(ctx context.Context, c table.Client, folder string) error {`
`1302`	`1303`	`options.WithPrimaryKeyColumn("series_id", "season_id"),`
`1303`	`1304`	`)`
`1304`	`1305`	`},`
	`1306`	`+ table.WithIdempotent(),`
`1305`	`1307`	`)`
`1306`	`1308`	`if err != nil {`
`1307`	`1309`	`return err`
`@@ -1322,6 +1324,7 @@ func createTables(ctx context.Context, c table.Client, folder string) error {`
`1322`	`1324`	`options.WithPrimaryKeyColumn("series_id", "season_id", "episode_id"),`
`1323`	`1325`	`)`
`1324`	`1326`	`},`
	`1327`	`+ table.WithIdempotent(),`
`1325`	`1328`	`)`
`1326`	`1329`	`return err`
`1327`	`1330`	`}`
`@@ -1399,6 +1402,7 @@ func TestLongStream(t *testing.T) {`
`1399`	`1402`	`CREATE TABLE `+tableName+` (val Int64, PRIMARY KEY (val))`,
`1400`	`1403`	`)`
`1401`	`1404`	`},`
	`1405`	`+ table.WithIdempotent(),`
`1402`	`1406`	`); err != nil {`
`1403`	`1407`	`t.Fatalf("create table failed: %v\n", err)`
`1404`	`1408`	`}`
`@@ -1454,6 +1458,7 @@ func TestLongStream(t *testing.T) {`
`1454`	`1458`	`)`
`1455`	`1459`	`return err`
`1456`	`1460`	`},`
	`1461`	`+ table.WithIdempotent(),`
`1457`	`1462`	`); err != nil {`
`1458`	`1463`	`t.Fatalf("upsert failed: %v\n", err)`
`1459`	`1464`	`} else {`
`@@ -1517,6 +1522,7 @@ func TestLongStream(t *testing.T) {`
`1517`	`1522`	`}`
`1518`	`1523`	`return nil`
`1519`	`1524`	`},`
	`1525`	`+ table.WithIdempotent(),`
`1520`	`1526`	`); err != nil {`
`1521`	`1527`	`t.Fatalf("stream query failed: %v\n", err)`
`1522`	`1528`	`}`
`@@ -1557,6 +1563,7 @@ func TestLongStream(t *testing.T) {`
`1557`	`1563`	`}`
`1558`	`1564`	`return nil`
`1559`	`1565`	`},`
	`1566`	`+ table.WithIdempotent(),`
`1560`	`1567`	`); err != nil {`
`1561`	`1568`	`t.Fatalf("stream query failed: %v\n", err)`
`1562`	`1569`	`}`
`@@ -1611,6 +1618,7 @@ func TestSplitRangesAndRead(t *testing.T) {`
`1611`	`1618`	)`,
`1612`	`1619`	`)`
`1613`	`1620`	`},`
	`1621`	`+ table.WithIdempotent(),`
`1614`	`1622`	`); err != nil {`
`1615`	`1623`	`t.Fatalf("create table failed: %v\n", err)`
`1616`	`1624`	`}`
`@@ -1666,6 +1674,7 @@ func TestSplitRangesAndRead(t *testing.T) {`
`1666`	`1674`	`)`
`1667`	`1675`	`return err`
`1668`	`1676`	`},`
	`1677`	`+ table.WithIdempotent(),`
`1669`	`1678`	`); err != nil {`
`1670`	`1679`	`t.Fatalf("upsert failed: %v\n", err)`
`1671`	`1680`	`} else {`
`@@ -1724,6 +1733,7 @@ func TestSplitRangesAndRead(t *testing.T) {`
`1724`	`1733`	`}`
`1725`	`1734`	`return nil`
`1726`	`1735`	`},`
	`1736`	`+ table.WithIdempotent(),`
`1727`	`1737`	`); err != nil {`
`1728`	`1738`	`t.Fatalf("stream query failed: %v\n", err)`
`1729`	`1739`	`}`
`@@ -1757,6 +1767,7 @@ func TestSplitRangesAndRead(t *testing.T) {`
`1757`	`1767`	`}`
`1758`	`1768`	`return nil`
`1759`	`1769`	`},`
	`1770`	`+ table.WithIdempotent(),`
`1760`	`1771`	`); err != nil {`
`1761`	`1772`	`t.Fatalf("stream query failed: %v\n", err)`
`1762`	`1773`	`}`
`@@ -1788,14 +1799,18 @@ func TestIssue229UnexpectedNullWhileParseNilJsonDocumentValue(t *testing.T) {`
`1788`	`1799`	`defer db.Close(ctx)`
`1789`	`1800`	`err := db.Table().DoTx(ctx, func(ctx context.Context, tx table.TransactionActor) error {`
`1790`	`1801`	res, err := tx.Execute(ctx, `SELECT Nothing(JsonDocument?) AS r`, nil)
`1791`		`- require.NoError(t, err)`
`1792`		`- require.NoError(t, res.NextResultSetErr(ctx))`
	`1802`	`+ if err != nil {`
	`1803`	`+ return err`
	`1804`	`+ }`
	`1805`	`+ if err = res.NextResultSetErr(ctx); err != nil {`
	`1806`	`+ return err`
	`1807`	`+ }`
`1793`	`1808`	`require.True(t, res.NextRow())`
`1794`	`1809`
`1795`	`1810`	`var val issue229Struct`
`1796`	`1811`	`require.NoError(t, res.Scan(&val))`
`1797`		`- return nil`
`1798`		`- })`
	`1812`	`+ return res.Err()`
	`1813`	`+ }, table.WithIdempotent())`
`1799`	`1814`	`require.NoError(t, err)`
`1800`	`1815`	`}`
`1801`	`1816`
`@@ -1822,8 +1837,12 @@ func TestIssue259IntervalFromDuration(t *testing.T) {`
`1822`	`1837`	SELECT $ts == $ten_micro, $ten_micro;`, table.NewQueryParameters(
`1823`	`1838`	table.ValueParam(`$ts`, types.IntervalValueFromDuration(10*time.Microsecond)),
`1824`	`1839`	`))`
`1825`		`- require.NoError(t, err)`
`1826`		`- require.NoError(t, res.NextResultSetErr(ctx))`
	`1840`	`+ if err != nil {`
	`1841`	`+ return err`
	`1842`	`+ }`
	`1843`	`+ if err = res.NextResultSetErr(ctx); err != nil {`
	`1844`	`+ return err`
	`1845`	`+ }`
`1827`	`1846`	`require.True(t, res.NextRow())`
`1828`	`1847`
`1829`	`1848`	`var (`
`@@ -1950,7 +1969,7 @@ func TestNullType(t *testing.T) {`
`1950`	`1969`	`}`
`1951`	`1970`	`fmt.Printf("%+v\n", rescheduleDue)`
`1952`	`1971`	`return res.Err()`
`1953`		`- }, table.WithTxSettings(table.TxSettings(table.WithSnapshotReadOnly())))`
	`1972`	`+ }, table.WithTxSettings(table.TxSettings(table.WithSnapshotReadOnly())), table.WithIdempotent())`
`1954`	`1973`	`require.NoError(t, err)`
`1955`	`1974`	`}`
`1956`	`1975`
`@@ -2193,7 +2212,7 @@ func TestValueToYqlLiteral(t *testing.T) {`
`2193`	`2212`	`require.Equal(t, 1, len(values))`
`2194`	`2213`	`require.Equal(t, tt.Yql(), values[0].Yql(), fmt.Sprintf("%T vs %T", tt, values[0]))`
`2195`	`2214`	`return nil`
`2196`		`- })`
	`2215`	`+ }, table.WithIdempotent())`
`2197`	`2216`	`require.NoError(t, err)`
`2198`	`2217`	`})`
`2199`	`2218`	`}`