适用于 .NET 的 Amazon QLDB 驱动程序 — 说明书参考 - Amazon Quantum Ledger Database (Amazon QLDB)

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

适用于 .NET 的 Amazon QLDB 驱动程序 — 说明书参考

本参考指南显示了 .NET 的 Amazon QLDB 驱动程序的常见用例。它提供了 C# 代码示例,演示了如何使用该驱动程序运行基本的创建、读取、更新和删除 (CRUD) 操作。它还包括用于处理 Amazon Ion 数据的代码示例。此外,本指南还重点介绍了使事务具有幂等性和实现唯一性约束的最佳实践。

注意

本主题提供了默认使用 Ion 对象映射器 处理 Amazon Ion 的代码示例。QLDB 在.NET 驱动程序的 1.3.0 版本中引入 Ion 对象映射器。在适用的情况下,本主题还提供了使用标准 Ion 库 作为替代方案代码示例。要了解更多信息,请参阅 使用 Amazon Ion

导入驱动程序

下面的代码示例导入驱动程序。

using Amazon.QLDB.Driver; using Amazon.QLDB.Driver.Generic; using Amazon.QLDB.Driver.Serialization;
using Amazon.QLDB.Driver; using Amazon.IonDotnet.Builders;

实例化驱动程序

以下代码示例使用默认设置创建连接到指定分类账名称的驱动程序实例。

Async
IAsyncQldbDriver driver = AsyncQldbDriver.Builder() .WithLedger("vehicle-registration") // Add Serialization library .WithSerializer(new ObjectSerializer()) .Build();
Sync
IQldbDriver driver = QldbDriver.Builder() .WithLedger("vehicle-registration") // Add Serialization library .WithSerializer(new ObjectSerializer()) .Build();
Async
IAsyncQldbDriver driver = AsyncQldbDriver.Builder().WithLedger("vehicle-registration").Build();
Sync
IQldbDriver driver = QldbDriver.Builder().WithLedger("vehicle-registration").Build();

CRUD 操作

QLDB 作为事务的一部分运行创建、读取、更新和删除 (CRUD) 操作。

警告

作为最佳实践,使写事务严格地幂等。

使事务幂等

我们建议将写事务设置为幂等,以避免重试时出现任何意想不到的副作用。如果事务可以运行多次并每次都产生相同的结果,则事务是幂等的

例如,假设有一个事务,要将文档插入名为 Person 的表中。事务应该首先检查文档是否已经存在于表中。如果没有这种检查,表最终可能会有重复的文档。

假设 QLDB 在服务器端成功提交了事务,但客户端在等待响应时超时。如果事务不是幂等的,则在重试的情况下可以多次插入相同的文档。

使用索引避免全表扫描

我们还建议您在索引字段或文档 ID 上使用相等运算符来运行带有WHERE谓词子句的语句;例如,WHERE indexedField = 123WHERE indexedField IN (456, 789)。如果没有这种索引查找,QLDB 需要进行表扫描,这可能会导致事务超时或乐观并发控制 (OCC) 冲突。

有关 OCC 的更多信息,请参阅Amazon QLDB 并发模型

隐式创建的事务

Amazon.QLDB.Driver.IQldbDriver.Execute 方法接受一个 lambda 函数,该函数接收一个 Amazon.QLDB.Driver.TransactionExecutor 的实例,您可以用它来运行语句。TransactionExecutor 的实例封装了隐式创建的事务。

您可以使用事务执行器的 Execute 执行方法在 lambda 函数中运行语句。当 lambda 函数返回时,驱动程序会隐式提交事务。

以下各节介绍如何运行基本的 CRUD 操作、指定自定义重试逻辑以及如何实现唯一性约束。

创建表

Async
IAsyncResult<Table> createResult = await driver.Execute(async txn => { IQuery<Table> query = txn.Query<Table>("CREATE TABLE Person"); return await txn.Execute(query); }); await foreach (var result in createResult) { Console.WriteLine("{ tableId: " + result.TableId + " }"); // The statement returns the created table ID: // { tableId: 4o5Uk09OcjC6PpJpLahceE } }
Sync
IResult<Table> createResult = driver.Execute( txn => { IQuery<Table> query = txn.Query<Table>("CREATE TABLE Person"); return txn.Execute(query); }); foreach (var result in createResult) { Console.WriteLine("{ tableId: " + result.TableId + " }"); // The statement returns the created table ID: // { tableId: 4o5Uk09OcjC6PpJpLahceE } }
Async
// The result from driver.Execute() is buffered into memory because once the // transaction is committed, streaming the result is no longer possible. IAsyncResult result = await driver.Execute(async txn => { return await txn.Execute("CREATE TABLE Person"); }); await foreach (IIonValue row in result) { Console.WriteLine(row.ToPrettyString()); // The statement returns the created table ID: // { // tableId: "4o5Uk09OcjC6PpJpLahceE" // } }
Sync
// The result from driver.Execute() is buffered into memory because once the // transaction is committed, streaming the result is no longer possible. IResult result = driver.Execute(txn => { return txn.Execute("CREATE TABLE Person"); }); foreach (IIonValue row in result) { Console.WriteLine(row.ToPrettyString()); // The statement returns the created table ID: // { // tableId: "4o5Uk09OcjC6PpJpLahceE" // } }

创建索引

Async
IAsyncResult<Table> createResult = await driver.Execute(async txn => { IQuery<Table> query = txn.Query<Table>("CREATE INDEX ON Person(firstName)"); return await txn.Execute(query); }); await foreach (var result in createResult) { Console.WriteLine("{ tableId: " + result.TableId + " }"); // The statement returns the updated table ID: // { tableId: 4o5Uk09OcjC6PpJpLahceE } }
Sync
IResult<Table> createResult = driver.Execute(txn => { IQuery<Table> query = txn.Query<Table>("CREATE INDEX ON Person(firstName)"); return txn.Execute(query); }); foreach (var result in createResult) { Console.WriteLine("{ tableId: " + result.TableId + " }"); // The statement returns the updated table ID: // { tableId: 4o5Uk09OcjC6PpJpLahceE } }
Async
IAsyncResult result = await driver.Execute(async txn => { return await txn.Execute("CREATE INDEX ON Person(GovId)"); }); await foreach (IIonValue row in result) { Console.WriteLine(row.ToPrettyString()); // The statement returns the updated table ID: // { // tableId: "4o5Uk09OcjC6PpJpLahceE" // } }
Sync
IResult result = driver.Execute(txn => { return txn.Execute("CREATE INDEX ON Person(GovId)"); }); foreach (IIonValue row in result) { Console.WriteLine(row.ToPrettyString()); // The statement returns the updated table ID: // { // tableId: "4o5Uk09OcjC6PpJpLahceE" // } }

阅读文档

// Assumes that Person table has documents as follows: // { "GovId": "TOYENC486FH", "FirstName" : "Brent" } // Person class is defined as follows: // public class Person // { // public string GovId { get; set; } // public string FirstName { get; set; } // } IAsyncResult<Person> result = await driver.Execute(async txn => { return await txn.Execute(txn.Query<Person>("SELECT * FROM Person WHERE GovId = 'TOYENC486FH'")); }); await foreach (Person person in result) { Console.WriteLine(person.GovId); // Prints TOYENC486FH. Console.WriteLine(person.FirstName); // Prints Brent. }
注意

当您在没有索引查找的情况下运行查询时,它会调用全表扫描。在此示例中,我们建议在GovId字段上设置 索引以优化性能。如果不开启GovId索引,查询可能会有更长的延迟,还可能导致 OCC 冲突异常或者事务超时。

使用查询参数

以下代码示例使用 C# 类型查询参数。

IAsyncResult<Person> result = await driver.Execute(async txn => { return await txn.Execute(txn.Query<Person>("SELECT * FROM Person WHERE FirstName = ?", "Brent")); }); await foreach (Person person in result) { Console.WriteLine(person.GovId); // Prints TOYENC486FH. Console.WriteLine(person.FirstName); // Prints Brent. }

以下代码示例使用了多个 C# 类型查询参数。

IAsyncResult<Person> result = await driver.Execute(async txn => { return await txn.Execute(txn.Query<Person>("SELECT * FROM Person WHERE GovId = ? AND FirstName = ?", "TOYENC486FH", "Brent")); }); await foreach (Person person in result) { Console.WriteLine(person.GovId); // Prints TOYENC486FH. Console.WriteLine(person.FirstName); // Prints Brent. }

以下代码示例使用 C# 类型查询参数。

// Assumes that Person table has documents as follows: // { "GovId": "TOYENC486FH", "FirstName" : "Brent" } // { "GovId": "ROEE1C1AABH", "FirstName" : "Jim" } // { "GovId": "YH844DA7LDB", "FirstName" : "Mary" } string[] ids = { "TOYENC486FH", "ROEE1C1AABH", "YH844DA7LDB" }; IAsyncResult<Person> result = await driver.Execute(async txn => { return await txn.Execute(txn.Query<Person>("SELECT * FROM Person WHERE GovId IN (?,?,?)", ids)); }); await foreach (Person person in result) { Console.WriteLine(person.FirstName); // Prints Brent on first iteration. // Prints Jim on second iteration. // Prints Mary on third iteration. }

以下代码示例使用 C# 列表为值。

// Assumes that Person table has document as follows: // { "GovId": "TOYENC486FH", // "FirstName" : "Brent", // "Vehicles": [ // { "Make": "Volkswagen", // "Model": "Golf"}, // { "Make": "Honda", // "Model": "Civic"} // ] // } // Person class is defined as follows: // public class Person // { // public string GovId { get; set; } // public string FirstName { get; set; } // public List<Vehicle> Vehicles { get; set; } // } // Vehicle class is defined as follows: // public class Vehicle // { // public string Make { get; set; } // public string Model { get; set; } // } List<Vehicle> vehicles = new List<Vehicle> { new Vehicle { Make = "Volkswagen", Model = "Golf" }, new Vehicle { Make = "Honda", Model = "Civic" } }; IAsyncResult<Person> result = await driver.Execute(async txn => { return await txn.Execute(txn.Query<Person>("SELECT * FROM Person WHERE Vehicles = ?", vehicles)); }); await foreach (Person person in result) { Console.WriteLine("{"); Console.WriteLine($" GovId: {person.GovId},"); Console.WriteLine($" FirstName: {person.FirstName},"); Console.WriteLine(" Vehicles: ["); foreach (Vehicle vehicle in person.Vehicles) { Console.WriteLine(" {"); Console.WriteLine($" Make: {vehicle.Make},"); Console.WriteLine($" Model: {vehicle.Model},"); Console.WriteLine(" },"); } Console.WriteLine(" ]"); Console.WriteLine("}"); // Prints: // { // GovId: TOYENC486FH, // FirstName: Brent, // Vehicles: [ // { // Make: Volkswagen, // Model: Golf // }, // { // Make: Honda, // Model: Civic // }, // ] // } }
Async
// Assumes that Person table has documents as follows: // { "GovId": "TOYENC486FH", "FirstName" : "Brent" } IAsyncResult result = await driver.Execute(async txn => { return await txn.Execute("SELECT * FROM Person WHERE GovId = 'TOYENC486FH'"); }); await foreach (IIonValue row in result) { Console.WriteLine(row.GetField("GovId").StringValue); // Prints TOYENC486FH. Console.WriteLine(row.GetField("FirstName").StringValue); // Prints Brent. }
Sync
// Assumes that Person table has documents as follows: // { "GovId": "TOYENC486FH", "FirstName" : "Brent" } IResult result = driver.Execute(txn => { return txn.Execute("SELECT * FROM Person WHERE GovId = 'TOYENC486FH'"); }); foreach (IIonValue row in result) { Console.WriteLine(row.GetField("GovId").StringValue); // Prints TOYENC486FH. Console.WriteLine(row.GetField("FirstName").StringValue); // Prints Brent. }
注意

当您在没有索引查找的情况下运行查询时,它会调用全表扫描。在此示例中,我们建议在GovId字段上设置 索引以优化性能。如果不开启GovId索引,查询可能会有更长的延迟,还可能导致 OCC 冲突异常或者事务超时。

以下代码示例使用 Ion 类型查询参数。

Async
IValueFactory valueFactory = new ValueFactory(); IIonValue ionFirstName = valueFactory.NewString("Brent"); IAsyncResult result = await driver.Execute(async txn => { return await txn.Execute("SELECT * FROM Person WHERE FirstName = ?", ionFirstName); }); await foreach (IIonValue row in result) { Console.WriteLine(row.GetField("GovId").StringValue); // Prints TOYENC486FH. Console.WriteLine(row.GetField("FirstName").StringValue); // Prints Brent. }
Sync
IValueFactory valueFactory = new ValueFactory(); IIonValue ionFirstName = valueFactory.NewString("Brent"); IResult result = driver.Execute(txn => { return txn.Execute("SELECT * FROM Person WHERE FirstName = ?", ionFirstName); }); foreach (IIonValue row in result) { Console.WriteLine(row.GetField("GovId").StringValue); // Prints TOYENC486FH. Console.WriteLine(row.GetField("FirstName").StringValue); // Prints Brent. }

以下代码示例使用了多个查询参数。

Async
IIonValue ionGovId = valueFactory.NewString("TOYENC486FH"); IIonValue ionFirstName = valueFactory.NewString("Brent"); IAsyncResult result = await driver.Execute(async txn => { return await txn.Execute("SELECT * FROM Person WHERE GovId = ? AND FirstName = ?", ionGovId, ionFirstName); }); await foreach (IIonValue row in result) { Console.WriteLine(row.GetField("GovId").StringValue); // Prints TOYENC486FH. Console.WriteLine(row.GetField("FirstName").StringValue); // Prints Brent. }
Sync
IIonValue ionGovId = valueFactory.NewString("TOYENC486FH"); IIonValue ionFirstName = valueFactory.NewString("Brent"); IResult result = driver.Execute(txn => { return txn.Execute("SELECT * FROM Person WHERE GovId = ? AND FirstName = ?", ionGovId, ionFirstName); }); foreach (IIonValue row in result) { Console.WriteLine(row.GetField("GovId").StringValue); // Prints TOYENC486FH. Console.WriteLine(row.GetField("FirstName").StringValue); // Prints Brent. }

以下代码示例使用查询参数列表。

Async
// Assumes that Person table has documents as follows: // { "GovId": "TOYENC486FH", "FirstName" : "Brent" } // { "GovId": "ROEE1C1AABH", "FirstName" : "Jim" } // { "GovId": "YH844DA7LDB", "FirstName" : "Mary" } IIonValue[] ionIds = { valueFactory.NewString("TOYENC486FH"), valueFactory.NewString("ROEE1C1AABH"), valueFactory.NewString("YH844DA7LDB") }; IAsyncResult result = await driver.Execute(async txn => { return await txn.Execute("SELECT * FROM Person WHERE GovId IN (?,?,?)", ionIds); }); await foreach (IIonValue row in result) { Console.WriteLine(row.GetField("FirstName").StringValue); // Prints Brent on first iteration. // Prints Jim on second iteration. // Prints Mary on third iteration. }
Sync
// Assumes that Person table has documents as follows: // { "GovId": "TOYENC486FH", "FirstName" : "Brent" } // { "GovId": "ROEE1C1AABH", "FirstName" : "Jim" } // { "GovId": "YH844DA7LDB", "FirstName" : "Mary" } IIonValue[] ionIds = { valueFactory.NewString("TOYENC486FH"), valueFactory.NewString("ROEE1C1AABH"), valueFactory.NewString("YH844DA7LDB") }; IResult result = driver.Execute(txn => { return txn.Execute("SELECT * FROM Person WHERE GovId IN (?,?,?)", ionIds); }); foreach (IIonValue row in result) { Console.WriteLine(row.GetField("FirstName").StringValue); // Prints Brent on first iteration. // Prints Jim on second iteration. // Prints Mary on third iteration. }

以下代码示例使用 C# 列表为值。要了解使用不同 Ion 类型任务的更多信息,请参阅 使用 Amazon QLDB 中的 Amazon Ion 数据类型

Async
// Assumes that Person table has document as follows: // { "GovId": "TOYENC486FH", // "FirstName" : "Brent", // "Vehicles": [ // { "Make": "Volkswagen", // "Model": "Golf"}, // { "Make": "Honda", // "Model": "Civic"} // ] // } IIonValue ionVehicle1 = valueFactory.NewEmptyStruct(); ionVehicle1.SetField("Make", valueFactory.NewString("Volkswagen")); ionVehicle1.SetField("Model", valueFactory.NewString("Golf")); IIonValue ionVehicle2 = valueFactory.NewEmptyStruct(); ionVehicle2.SetField("Make", valueFactory.NewString("Honda")); ionVehicle2.SetField("Model", valueFactory.NewString("Civic")); IIonValue ionVehicles = valueFactory.NewEmptyList(); ionVehicles.Add(ionVehicle1); ionVehicles.Add(ionVehicle2); IAsyncResult result = await driver.Execute(async txn => { return await txn.Execute("SELECT * FROM Person WHERE Vehicles = ?", ionVehicles); }); await foreach (IIonValue row in result) { Console.WriteLine(row.ToPrettyString()); // Prints: // { // GovId: "TOYENC486FN", // FirstName: "Brent", // Vehicles: [ // { // Make: "Volkswagen", // Model: "Golf" // }, // { // Make: "Honda", // Model: "Civic" // } // ] // } }
Sync
// Assumes that Person table has document as follows: // { "GovId": "TOYENC486FH", // "FirstName" : "Brent", // "Vehicles": [ // { "Make": "Volkswagen", // "Model": "Golf"}, // { "Make": "Honda", // "Model": "Civic"} // ] // } IIonValue ionVehicle1 = valueFactory.NewEmptyStruct(); ionVehicle1.SetField("Make", valueFactory.NewString("Volkswagen")); ionVehicle1.SetField("Model", valueFactory.NewString("Golf")); IIonValue ionVehicle2 = valueFactory.NewEmptyStruct(); ionVehicle2.SetField("Make", valueFactory.NewString("Honda")); ionVehicle2.SetField("Model", valueFactory.NewString("Civic")); IIonValue ionVehicles = valueFactory.NewEmptyList(); ionVehicles.Add(ionVehicle1); ionVehicles.Add(ionVehicle2); IResult result = driver.Execute(txn => { return txn.Execute("SELECT * FROM Person WHERE Vehicles = ?", ionVehicles); }); foreach (IIonValue row in result) { Console.WriteLine(row.ToPrettyString()); // Prints: // { // GovId: "TOYENC486FN", // FirstName: "Brent", // Vehicles: [ // { // Make: "Volkswagen", // Model: "Golf" // }, // { // Make: "Honda", // Model: "Civic" // } // ] // } }

插入文档

以下代码示例插入 Ion 数据类型。

string govId = "TOYENC486FH"; Person person = new Person { GovId = "TOYENC486FH", FirstName = "Brent" }; await driver.Execute(async txn => { // Check if a document with GovId:TOYENC486FH exists // This is critical to make this transaction idempotent IAsyncResult<Person> result = await txn.Execute(txn.Query<Person>("SELECT * FROM Person WHERE GovId = ?", govId)); // Check if there is a record in the cursor. int count = await result.CountAsync(); if (count > 0) { // Document already exists, no need to insert return; } // Insert the document. await txn.Execute(txn.Query<Document>("INSERT INTO Person ?", person)); });
Async
IIonValue ionGovId = valueFactory.NewString("TOYENC486FH"); IIonValue ionPerson = valueFactory.NewEmptyStruct(); ionPerson.SetField("GovId", valueFactory.NewString("TOYENC486FH")); ionPerson.SetField("FirstName", valueFactory.NewString("Brent")); await driver.Execute(async txn => { // Check if a document with GovId:TOYENC486FH exists // This is critical to make this transaction idempotent IAsyncResult result = await txn.Execute("SELECT * FROM Person WHERE GovId = ?", ionGovId); // Check if there is a record in the cursor. int count = await result.CountAsync(); if (count > 0) { // Document already exists, no need to insert return; } // Insert the document. await txn.Execute("INSERT INTO Person ?", ionPerson); });
Sync
IIonValue ionGovId = valueFactory.NewString("TOYENC486FH"); IIonValue ionPerson = valueFactory.NewEmptyStruct(); ionPerson.SetField("GovId", valueFactory.NewString("TOYENC486FH")); ionPerson.SetField("FirstName", valueFactory.NewString("Brent")); driver.Execute(txn => { // Check if a document with GovId:TOYENC486FH exists // This is critical to make this transaction idempotent IResult result = txn.Execute("SELECT * FROM Person WHERE GovId = ?", ionGovId); // Check if there is a record in the cursor. int count = result.Count(); if (count > 0) { // Document already exists, no need to insert return; } // Insert the document. txn.Execute("INSERT INTO Person ?", ionPerson); });

此事务将文档插入 Person 表中。在插入之前,它首先检查文档是否已存在于表格内。此检查使事务本质上是幂等。即使您多次运行此事务,也不会造成任何异常副作用。

注意

在此示例中,我们建议在 GovId 字段上设置索引以优化性能。如果不开启GovId索引,语句可能会有更长的延迟,还可能导致 OCC 冲突异常或者事务超时。

在一条语句内插入多个文档

要使用单个 INSERT 语句插入多个文档,可以按如下方式向该语句传递 C# List 参数。

Person person1 = new Person { FirstName = "Brent", GovId = "TOYENC486FH" }; Person person2 = new Person { FirstName = "Jim", GovId = "ROEE1C1AABH" }; List<Person> people = new List<Person>(); people.Add(person1); people.Add(person2); IAsyncResult<Document> result = await driver.Execute(async txn => { return await txn.Execute(txn.Query<Document>("INSERT INTO Person ?", people)); }); await foreach (Document row in result) { Console.WriteLine("{ documentId: " + row.DocumentId + " }"); // The statement returns the created documents' ID: // { documentId: 6BFt5eJQDFLBW2aR8LPw42 } // { documentId: K5Zrcb6N3gmIEHgGhwoyKF } }

要使用单个 INSERT 语句插入多个文档,可以向该语句传递一个Ion 列表类型的参数,如下所示。

Async
IIonValue ionPerson1 = valueFactory.NewEmptyStruct(); ionPerson1.SetField("FirstName", valueFactory.NewString("Brent")); ionPerson1.SetField("GovId", valueFactory.NewString("TOYENC486FH")); IIonValue ionPerson2 = valueFactory.NewEmptyStruct(); ionPerson2.SetField("FirstName", valueFactory.NewString("Jim")); ionPerson2.SetField("GovId", valueFactory.NewString("ROEE1C1AABH")); IIonValue ionPeople = valueFactory.NewEmptyList(); ionPeople.Add(ionPerson1); ionPeople.Add(ionPerson2); IAsyncResult result = await driver.Execute(async txn => { return await txn.Execute("INSERT INTO Person ?", ionPeople); }); await foreach (IIonValue row in result) { Console.WriteLine(row.ToPrettyString()); // The statement returns the created documents' ID: // { // documentId: "6BFt5eJQDFLBW2aR8LPw42" // } // // { // documentId: "K5Zrcb6N3gmIEHgGhwoyKF" // } }
Sync
IIonValue ionPerson1 = valueFactory.NewEmptyStruct(); ionPerson1.SetField("FirstName", valueFactory.NewString("Brent")); ionPerson1.SetField("GovId", valueFactory.NewString("TOYENC486FH")); IIonValue ionPerson2 = valueFactory.NewEmptyStruct(); ionPerson2.SetField("FirstName", valueFactory.NewString("Jim")); ionPerson2.SetField("GovId", valueFactory.NewString("ROEE1C1AABH")); IIonValue ionPeople = valueFactory.NewEmptyList(); ionPeople.Add(ionPerson1); ionPeople.Add(ionPerson2); IResult result = driver.Execute(txn => { return txn.Execute("INSERT INTO Person ?", ionPeople); }); foreach (IIonValue row in result) { Console.WriteLine(row.ToPrettyString()); // The statement returns the created documents' ID: // { // documentId: "6BFt5eJQDFLBW2aR8LPw42" // } // // { // documentId: "K5Zrcb6N3gmIEHgGhwoyKF" // } }

传递 Ion 列表时,不要将变量占位符 (?) 括在双尖括号 (<<...>>) 内。在手动 PartiQL 语句中,双尖括号表示名为bag的无序集合。

更新文档

string govId = "TOYENC486FH"; string firstName = "John"; IAsyncResult<Document> result = await driver.Execute(async txn => { return await txn.Execute(txn.Query<Document>("UPDATE Person SET FirstName = ? WHERE GovId = ?", firstName , govId)); }); await foreach (Document row in result) { Console.WriteLine("{ documentId: " + row.DocumentId + " }"); // The statement returns the updated document ID: // { documentId: Djg30Zoltqy5M4BFsA2jSJ } }
Async
IIonValue ionGovId = valueFactory.NewString("TOYENC486FH"); IIonValue ionFirstName = valueFactory.NewString("John"); IAsyncResult result = await driver.Execute(async txn => { return await txn.Execute("UPDATE Person SET FirstName = ? WHERE GovId = ?", ionFirstName , ionGovId); }); await foreach (IIonValue row in result) { Console.WriteLine(row.ToPrettyString()); // The statement returns the updated document ID: // { // documentId: "Djg30Zoltqy5M4BFsA2jSJ" // } }
Sync
IIonValue ionGovId = valueFactory.NewString("TOYENC486FH"); IIonValue ionFirstName = valueFactory.NewString("John"); IResult result = driver.Execute(txn => { return txn.Execute("UPDATE Person SET FirstName = ? WHERE GovId = ?", ionFirstName , ionGovId); }); foreach (IIonValue row in result) { Console.WriteLine(row.ToPrettyString()); // The statement returns the updated document ID: // { // documentId: "Djg30Zoltqy5M4BFsA2jSJ" // } }
注意

在此示例中,我们建议在 GovId 字段上设置索引以优化性能。如果不开启GovId索引,语句可能会有更长的延迟,还可能导致 OCC 冲突异常或者事务超时。

删除文档

string govId = "TOYENC486FH"; IAsyncResult<Document> result = await driver.Execute(async txn => { return await txn.Execute(txn.Query<Document>("DELETE FROM Person WHERE GovId = ?", govId)); }); await foreach (Document row in result) { Console.WriteLine("{ documentId: " + row.DocumentId + " }"); // The statement returns the updated document ID: // { documentId: Djg30Zoltqy5M4BFsA2jSJ } }
Async
IIonValue ionGovId = valueFactory.NewString("TOYENC486FH"); IAsyncResult result = await driver.Execute(async txn => { return await txn.Execute("DELETE FROM Person WHERE GovId = ?", ionGovId); }); await foreach (IIonValue row in result) { Console.WriteLine(row.ToPrettyString()); // The statement returns the deleted document ID: // { // documentId: "Djg30Zoltqy5M4BFsA2jSJ" // } }
Sync
IIonValue ionGovId = valueFactory.NewString("TOYENC486FH"); IResult result = driver.Execute(txn => { return txn.Execute("DELETE FROM Person WHERE GovId = ?", ionGovId); }); foreach (IIonValue row in result) { Console.WriteLine(row.ToPrettyString()); // The statement returns the deleted document ID: // { // documentId: "Djg30Zoltqy5M4BFsA2jSJ" // } }
注意

在此示例中,我们建议在 GovId 字段上设置索引以优化性能。如果不开启GovId索引,语句可能会有更长的延迟,还可能导致 OCC 冲突异常或者事务超时。

在一个事务中运行多条语句

// This code snippet is intentionally trivial. In reality you wouldn't do this because you'd // set your UPDATE to filter on vin and insured, and check if you updated something or not. public static async Task<bool> InsureVehicle(IAsyncQldbDriver driver, string vin) { return await driver.Execute(async txn => { // Check if the vehicle is insured. Amazon.QLDB.Driver.Generic.IAsyncResult<Vehicle> result = await txn.Execute( txn.Query<Vehicle>("SELECT insured FROM Vehicles WHERE vin = ? AND insured = FALSE", vin)); if (await result.CountAsync() > 0) { // If the vehicle is not insured, insure it. await txn.Execute( txn.Query<Document>("UPDATE Vehicles SET insured = TRUE WHERE vin = ?", vin)); return true; } return false; }); }
Async
// This code snippet is intentionally trivial. In reality you wouldn't do this because you'd // set your UPDATE to filter on vin and insured, and check if you updated something or not. public static async Task<bool> InsureVehicle(IAsyncQldbDriver driver, string vin) { ValueFactory valueFactory = new ValueFactory(); IIonValue ionVin = valueFactory.NewString(vin); return await driver.Execute(async txn => { // Check if the vehicle is insured. Amazon.QLDB.Driver.IAsyncResult result = await txn.Execute( "SELECT insured FROM Vehicles WHERE vin = ? AND insured = FALSE", ionVin); if (await result.CountAsync() > 0) { // If the vehicle is not insured, insure it. await txn.Execute( "UPDATE Vehicles SET insured = TRUE WHERE vin = ?", ionVin); return true; } return false; }); }

重试逻辑

有关驱动程序的内置重试逻辑的信息,请参见 使用 Amazon QLDB 中的驱动程序了解重试策略

实现唯一限制

QLDB 不支持唯一索引,但您可在应用程序中实现此行为。

假设您要对 Person 表中的 GovId 字段实现唯一性约束。据此,可以编写执行以下操作的事务:

  1. 断言该表中没有指定 GovId 的现有文档。

  2. 如果断言通过,请插入文档。

如果一个竞争事务同时通过断言,则只有一个事务将成功提交。另一笔事务将失败,并显示 OCC 冲突异常。

以下代码示例显示了如何实现此唯一约束。

string govId = "TOYENC486FH"; Person person = new Person { GovId = "TOYENC486FH", FirstName = "Brent" }; await driver.Execute(async txn => { // Check if a document with GovId:TOYENC486FH exists // This is critical to make this transaction idempotent IAsyncResult<Person> result = await txn.Execute(txn.Query<Person>("SELECT * FROM Person WHERE GovId = ?", govId)); // Check if there is a record in the cursor. int count = await result.CountAsync(); if (count > 0) { // Document already exists, no need to insert return; } // Insert the document. await txn.Execute(txn.Query<Document>("INSERT INTO Person ?", person)); });
Async
IIonValue ionGovId = valueFactory.NewString("TOYENC486FH"); IIonValue ionPerson = valueFactory.NewEmptyStruct(); ionPerson.SetField("GovId", valueFactory.NewString("TOYENC486FH")); ionPerson.SetField("FirstName", valueFactory.NewString("Brent")); await driver.Execute(async txn => { // Check if a document with GovId:TOYENC486FH exists // This is critical to make this transaction idempotent IAsyncResult result = await txn.Execute("SELECT * FROM Person WHERE GovId = ?", ionGovId); // Check if there is a record in the cursor. int count = await result.CountAsync(); if (count > 0) { // Document already exists, no need to insert return; } // Insert the document. await txn.Execute("INSERT INTO Person ?", ionPerson); });
Sync
IIonValue ionGovId = valueFactory.NewString("TOYENC486FH"); IIonValue ionPerson = valueFactory.NewEmptyStruct(); ionPerson.SetField("GovId", valueFactory.NewString("TOYENC486FH")); ionPerson.SetField("FirstName", valueFactory.NewString("Brent")); driver.Execute(txn => { // Check if a document with GovId:TOYENC486FH exists // This is critical to make this transaction idempotent IResult result = txn.Execute("SELECT * FROM Person WHERE GovId = ?", ionGovId); // Check if there is a record in the cursor. int count = result.Count(); if (count > 0) { // Document already exists, no need to insert return; } // Insert the document. txn.Execute("INSERT INTO Person ?", ionPerson); });
注意

在此示例中,我们建议在 GovId 字段上设置索引以优化性能。如果不开启GovId索引,语句可能会有更长的延迟,还可能导致 OCC 冲突异常或者事务超时。

使用 Amazon Ion

可通过多种方法在 QLDB 中处理 Amazon Ion 数据。您可使用 Ion 库创建和修改 Ion 值。或者,您可以使用 Ion 对象映射器将 C# 普通旧 CLR 对象 (POCO) 映射进出 Ion 值。适用 .NET 的 Amazon QLDB 驱动程序 1.3 版本引入了对 Ion 对象映射器的支持。

以下各节提供了使用这两种技术处理 Ion 数据的代码示例。

导入 Ion 模块

using Amazon.IonObjectMapper;
using Amazon.IonDotnet.Builders;

创建 Ion 类型

以下代码示例介绍了如何使用 Ion 对象映射器从 C# 对象创建 Ion 值。

// Assumes that Person class is defined as follows: // public class Person // { // public string FirstName { get; set; } // public int Age { get; set; } // } // Initialize the Ion Object Mapper IonSerializer ionSerializer = new IonSerializer(); // The C# object to be serialized Person person = new Person { FirstName = "John", Age = 13 }; // Serialize the C# object into stream using the Ion Object Mapper Stream stream = ionSerializer.Serialize(person); // Load will take in stream and return a datagram; a top level container of Ion values. IIonValue ionDatagram = IonLoader.Default.Load(stream); // To get the Ion value within the datagram, we call GetElementAt(0). IIonValue ionPerson = ionDatagram.GetElementAt(0); Console.WriteLine(ionPerson.GetField("firstName").StringValue); Console.WriteLine(ionPerson.GetField("age").IntValue);

以下代码示例介绍了使用 Ion 库创建 Ion 值的两种方法。

使用 ValueFactory

using Amazon.IonDotnet.Tree; using Amazon.IonDotnet.Tree.Impl; IValueFactory valueFactory = new ValueFactory(); IIonValue ionPerson = valueFactory.NewEmptyStruct(); ionPerson.SetField("firstName", valueFactory.NewString("John")); ionPerson.SetField("age", valueFactory.NewInt(13)); Console.WriteLine(ionPerson.GetField("firstName").StringValue); Console.WriteLine(ionPerson.GetField("age").IntValue);

使用 IonLoader

using Amazon.IonDotnet.Builders; using Amazon.IonDotnet.Tree; // Load will take in Ion text and return a datagram; a top level container of Ion values. IIonValue ionDatagram = IonLoader.Default.Load("{firstName: \"John\", age: 13}"); // To get the Ion value within the datagram, we call GetElementAt(0). IIonValue ionPerson = ionDatagram.GetElementAt(0); Console.WriteLine(ionPerson.GetField("firstName").StringValue); Console.WriteLine(ionPerson.GetField("age").IntValue);

获取 Ion 二进制转储

// Initialize the Ion Object Mapper with Ion binary serialization format IonSerializer ionSerializer = new IonSerializer(new IonSerializationOptions { Format = IonSerializationFormat.BINARY }); // The C# object to be serialized Person person = new Person { FirstName = "John", Age = 13 }; MemoryStream stream = (MemoryStream) ionSerializer.Serialize(person); Console.WriteLine(BitConverter.ToString(stream.ToArray()));
// ionObject is an Ion struct MemoryStream stream = new MemoryStream(); using (var writer = IonBinaryWriterBuilder.Build(stream)) { ionObject.WriteTo(writer); writer.Finish(); } Console.WriteLine(BitConverter.ToString(stream.ToArray()));

获取 Ion 文本转储

// Initialize the Ion Object Mapper IonSerializer ionSerializer = new IonSerializer(new IonSerializationOptions { Format = IonSerializationFormat.TEXT }); // The C# object to be serialized Person person = new Person { FirstName = "John", Age = 13 }; MemoryStream stream = (MemoryStream) ionSerializer.Serialize(person); Console.WriteLine(System.Text.Encoding.UTF8.GetString(stream.ToArray()));
// ionObject is an Ion struct StringWriter sw = new StringWriter(); using (var writer = IonTextWriterBuilder.Build(sw)) { ionObject.WriteTo(writer); writer.Finish(); } Console.WriteLine(sw.ToString());

有关使用 Ion 的更多信息,请参阅 GitHub 上的 Amazon Ion 文档。有关在 QLDB 中使用 Ion 的更多代码示例,请参阅使用 Amazon QLDB 中的 Amazon Ion 数据类型