-
Notifications
You must be signed in to change notification settings - Fork 107
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.
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
docs: add samples for PostgresSQL #1781
Changes from 11 commits
376e449
db81ea6
5301349
e764e9d
332f15d
45b1358
9d6b60e
2759e7a
c4c2389
daede6b
0a50828
1c900f1
98a82e7
04eaa97
c8a9ccf
64d87ab
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,90 @@ | ||
/* | ||
* Copyright 2022 Google Inc. | ||
* | ||
* Licensed under the Apache License, Version 2.0 (the "License"); | ||
* you may not use this file except in compliance with the License. | ||
* You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package com.example.spanner; | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Would it make more sense to place all the PG samples in a separate sub-package. So something like There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Sure, will decide in today call with @jin-jj and decide the approach. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @olavloite : I think this was templated on https://github.com/googleapis/java-spanner/pull/1700/files. IIRC, If we move to a sub-package - the class path would end up becoming something like The seggregation still happens in the current scheme of things, and looking for Spangres samples with an explicit pg in them sounds all right to me. So we can keep things as is. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Thanks @olavloite and @ansh0l. So @Sivakumar-Searce we will keep the class name as-is. |
||
|
||
// [START spanner_postgresql_async_query_to_list] | ||
import com.google.api.core.ApiFuture; | ||
import com.google.cloud.spanner.AsyncResultSet; | ||
import com.google.cloud.spanner.DatabaseClient; | ||
import com.google.cloud.spanner.DatabaseId; | ||
import com.google.cloud.spanner.Spanner; | ||
import com.google.cloud.spanner.SpannerOptions; | ||
import com.google.cloud.spanner.Statement; | ||
import java.util.List; | ||
import java.util.concurrent.ExecutionException; | ||
import java.util.concurrent.ExecutorService; | ||
import java.util.concurrent.Executors; | ||
import java.util.concurrent.TimeUnit; | ||
import java.util.concurrent.TimeoutException; | ||
|
||
public class PgAsyncQueryToListAsyncExample { | ||
static class Album { | ||
final long singerId; | ||
final long albumId; | ||
final String albumTitle; | ||
|
||
Album(long singerId, long albumId, String albumTitle) { | ||
this.singerId = singerId; | ||
this.albumId = albumId; | ||
this.albumTitle = albumTitle; | ||
} | ||
} | ||
|
||
static void asyncQueryToList() throws InterruptedException, ExecutionException, TimeoutException { | ||
// TODO(developer): Replace these variables before running the sample. | ||
String projectId = "my-project"; | ||
String instanceId = "my-instance"; | ||
String databaseId = "my-database"; | ||
|
||
try (Spanner spanner = | ||
SpannerOptions.newBuilder().setProjectId(projectId).build().getService()) { | ||
DatabaseClient client = | ||
spanner.getDatabaseClient(DatabaseId.of(projectId, instanceId, databaseId)); | ||
asyncQueryToList(client); | ||
} | ||
} | ||
|
||
// Execute a query asynchronously and transform the result to a list. | ||
static void asyncQueryToList(DatabaseClient client) | ||
throws InterruptedException, ExecutionException, TimeoutException { | ||
ExecutorService executor = Executors.newSingleThreadExecutor(); | ||
ApiFuture<? extends List<Album>> albums; | ||
try (AsyncResultSet resultSet = | ||
client | ||
.singleUse() | ||
.executeQueryAsync(Statement.of("SELECT singerid as \"SingerId\", " | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The same as the comments, I am thinking renaming all column name in DDL from SingerId to singer_id, which is more pg friendly. I will circle back later to update what should we do. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We will keep the code as-is. No change required. |
||
+ "albumid as \"AlbumId\", albumtitle as \"AlbumTitle\" " | ||
+ "FROM Albums"))) { | ||
// Convert the result set to a list of Albums asynchronously. | ||
albums = | ||
resultSet.toListAsync( | ||
reader -> { | ||
return new Album( | ||
reader.getLong("SingerId"), | ||
reader.getLong("AlbumId"), | ||
reader.getString("AlbumTitle")); | ||
}, | ||
executor); | ||
} | ||
|
||
for (Album album : albums.get(30L, TimeUnit.SECONDS)) { | ||
System.out.printf("%d %d %s%n", album.singerId, album.albumId, album.albumTitle); | ||
} | ||
executor.shutdown(); | ||
} | ||
} | ||
//[END spanner_postgresql_async_query_to_list] |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,124 @@ | ||
/* | ||
* Copyright 2022 Google Inc. | ||
* | ||
* Licensed under the Apache License, Version 2.0 (the "License"); | ||
* you may not use this file except in compliance with the License. | ||
* You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package com.example.spanner; | ||
|
||
//[START spanner_postgresql_async_read_write_transaction] | ||
import com.google.api.core.ApiFuture; | ||
import com.google.api.core.ApiFutures; | ||
import com.google.cloud.spanner.AsyncRunner; | ||
import com.google.cloud.spanner.DatabaseClient; | ||
import com.google.cloud.spanner.DatabaseId; | ||
import com.google.cloud.spanner.Key; | ||
import com.google.cloud.spanner.Spanner; | ||
import com.google.cloud.spanner.SpannerExceptionFactory; | ||
import com.google.cloud.spanner.SpannerOptions; | ||
import com.google.cloud.spanner.Statement; | ||
import com.google.cloud.spanner.Struct; | ||
import com.google.common.collect.ImmutableList; | ||
import com.google.common.util.concurrent.MoreExecutors; | ||
import java.util.Arrays; | ||
import java.util.concurrent.ExecutionException; | ||
import java.util.concurrent.ExecutorService; | ||
import java.util.concurrent.Executors; | ||
import java.util.concurrent.TimeUnit; | ||
import java.util.concurrent.TimeoutException; | ||
|
||
class PgAsyncRunnerExample { | ||
|
||
static void asyncRunner() throws InterruptedException, ExecutionException, TimeoutException { | ||
// TODO(developer): Replace these variables before running the sample. | ||
String projectId = "my-project"; | ||
String instanceId = "my-instance"; | ||
String databaseId = "my-database"; | ||
|
||
try (Spanner spanner = | ||
SpannerOptions.newBuilder().setProjectId(projectId).build().getService()) { | ||
DatabaseClient client = | ||
spanner.getDatabaseClient(DatabaseId.of(projectId, instanceId, databaseId)); | ||
asyncRunner(client); | ||
} | ||
} | ||
|
||
// Execute a read/write transaction asynchronously. | ||
static void asyncRunner(DatabaseClient client) | ||
throws InterruptedException, ExecutionException, TimeoutException { | ||
ExecutorService executor = Executors.newSingleThreadExecutor(); | ||
|
||
// Create an async transaction runner. | ||
AsyncRunner runner = client.runAsync(); | ||
// The transaction returns the total number of rows that were updated as a future array of | ||
// longs. | ||
ApiFuture<long[]> rowCounts = | ||
runner.runAsync( | ||
txn -> { | ||
// Transfer marketing budget from one album to another. We do it in a | ||
// transaction to ensure that the transfer is atomic. | ||
ApiFuture<Struct> album1BudgetFut = | ||
txn.readRowAsync("Albums", Key.of(1, 1), ImmutableList.of("MarketingBudget")); | ||
ApiFuture<Struct> album2BudgetFut = | ||
txn.readRowAsync("Albums", Key.of(2, 2), ImmutableList.of("MarketingBudget")); | ||
|
||
try { | ||
// Transaction will only be committed if this condition still holds at the | ||
// time of commit. Otherwise it will be aborted and the AsyncWork will be | ||
// rerun by the client library. | ||
long transfer = 200_000; | ||
if (album2BudgetFut.get().getLong(0) >= transfer) { | ||
long album1Budget = album1BudgetFut.get().getLong(0); | ||
long album2Budget = album2BudgetFut.get().getLong(0); | ||
|
||
album1Budget += transfer; | ||
album2Budget -= transfer; | ||
Statement updateStatement1 = | ||
Statement.newBuilder( | ||
"UPDATE Albums " | ||
+ "SET MarketingBudget = $1 " | ||
+ "WHERE SingerId = 1 and AlbumId = 1") | ||
.bind("p1") | ||
.to(album1Budget) | ||
.build(); | ||
Statement updateStatement2 = | ||
Statement.newBuilder( | ||
"UPDATE Albums " | ||
+ "SET MarketingBudget = $1 " | ||
+ "WHERE SingerId = 2 and AlbumId = 2") | ||
.bind("p1") | ||
.to(album2Budget) | ||
.build(); | ||
return txn.batchUpdateAsync( | ||
ImmutableList.of(updateStatement1, updateStatement2)); | ||
} else { | ||
return ApiFutures.immediateFuture(new long[] {0L, 0L}); | ||
} | ||
} catch (ExecutionException e) { | ||
throw SpannerExceptionFactory.newSpannerException(e.getCause()); | ||
} catch (InterruptedException e) { | ||
throw SpannerExceptionFactory.propagateInterrupt(e); | ||
} | ||
}, | ||
executor); | ||
|
||
ApiFuture<Long> totalUpdateCount = | ||
ApiFutures.transform( | ||
rowCounts, | ||
input -> Arrays.stream(input).sum(), | ||
MoreExecutors.directExecutor()); | ||
System.out.printf("%d records updated.%n", totalUpdateCount.get(30L, TimeUnit.SECONDS)); | ||
executor.shutdown(); | ||
} | ||
} | ||
//[END spanner_postgresql_async_read_write_transaction] |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,140 @@ | ||
/* | ||
* Copyright 2022 Google Inc. | ||
* | ||
* Licensed under the Apache License, Version 2.0 (the "License"); | ||
* you may not use this file except in compliance with the License. | ||
* You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package com.example.spanner; | ||
|
||
//[START spanner_postgresql_async_transaction_manager] | ||
import com.google.api.core.ApiFuture; | ||
import com.google.api.core.ApiFutures; | ||
import com.google.cloud.spanner.AbortedException; | ||
import com.google.cloud.spanner.AsyncTransactionManager; | ||
import com.google.cloud.spanner.AsyncTransactionManager.AsyncTransactionStep; | ||
import com.google.cloud.spanner.AsyncTransactionManager.CommitTimestampFuture; | ||
import com.google.cloud.spanner.AsyncTransactionManager.TransactionContextFuture; | ||
import com.google.cloud.spanner.DatabaseClient; | ||
import com.google.cloud.spanner.DatabaseId; | ||
import com.google.cloud.spanner.Key; | ||
import com.google.cloud.spanner.Spanner; | ||
import com.google.cloud.spanner.SpannerOptions; | ||
import com.google.cloud.spanner.Statement; | ||
import com.google.cloud.spanner.Struct; | ||
import com.google.common.collect.ImmutableList; | ||
import com.google.common.util.concurrent.MoreExecutors; | ||
import java.util.Arrays; | ||
import java.util.List; | ||
import java.util.concurrent.ExecutionException; | ||
import java.util.concurrent.ExecutorService; | ||
import java.util.concurrent.Executors; | ||
import java.util.concurrent.TimeUnit; | ||
import java.util.concurrent.TimeoutException; | ||
|
||
class PgAsyncTransactionManagerExample { | ||
|
||
static void asyncTransactionManager() | ||
throws InterruptedException, ExecutionException, TimeoutException { | ||
// TODO(developer): Replace these variables before running the sample. | ||
String projectId = "my-project"; | ||
String instanceId = "my-instance"; | ||
String databaseId = "my-database"; | ||
|
||
try (Spanner spanner = | ||
SpannerOptions.newBuilder().setProjectId(projectId).build().getService()) { | ||
DatabaseClient client = | ||
spanner.getDatabaseClient(DatabaseId.of(projectId, instanceId, databaseId)); | ||
asyncTransactionManager(client); | ||
} | ||
} | ||
|
||
static void asyncTransactionManager(DatabaseClient client) | ||
throws InterruptedException, ExecutionException, TimeoutException { | ||
ExecutorService executor = Executors.newSingleThreadExecutor(); | ||
|
||
AsyncTransactionStep<List<Struct>, long[]> updateCounts; | ||
try (AsyncTransactionManager mgr = client.transactionManagerAsync()) { | ||
TransactionContextFuture txn = mgr.beginAsync(); | ||
// Loop to retry aborted errors. | ||
while (true) { | ||
try { | ||
updateCounts = | ||
txn.then( | ||
(transaction, v) -> { | ||
// Execute two reads in parallel and return the result of these as the input | ||
// for the next step of the transaction. | ||
ApiFuture<Struct> album1BudgetFut = | ||
transaction.readRowAsync( | ||
"Albums", Key.of(1, 1), ImmutableList.of("MarketingBudget")); | ||
ApiFuture<Struct> album2BudgetFut = | ||
transaction.readRowAsync( | ||
"Albums", Key.of(2, 2), ImmutableList.of("MarketingBudget")); | ||
return ApiFutures.allAsList( | ||
Arrays.asList(album1BudgetFut, album2BudgetFut)); | ||
}, | ||
executor) | ||
// The input of the next step of the transaction is the return value of the | ||
// previous step, i.e. a list containing the marketing budget of two Albums. | ||
.then( | ||
(transaction, budgets) -> { | ||
long album1Budget = budgets.get(0).getLong(0); | ||
long album2Budget = budgets.get(1).getLong(0); | ||
long transfer = 200_000; | ||
if (album2Budget >= transfer) { | ||
album1Budget += transfer; | ||
album2Budget -= transfer; | ||
Statement updateStatement1 = | ||
Statement.newBuilder( | ||
"UPDATE Albums " | ||
+ "SET MarketingBudget = $1 " | ||
+ "WHERE SingerId = 1 and AlbumId = 1") | ||
.bind("p1") | ||
.to(album1Budget) | ||
.build(); | ||
Statement updateStatement2 = | ||
Statement.newBuilder( | ||
"UPDATE Albums " | ||
+ "SET MarketingBudget = $1 " | ||
+ "WHERE SingerId = 2 and AlbumId = 2") | ||
.bind("p1") | ||
.to(album2Budget) | ||
.build(); | ||
return transaction.batchUpdateAsync( | ||
ImmutableList.of(updateStatement1, updateStatement2)); | ||
} else { | ||
return ApiFutures.immediateFuture(new long[] {0L, 0L}); | ||
} | ||
}, | ||
executor); | ||
// Commit after the updates. | ||
CommitTimestampFuture commitTsFut = updateCounts.commitAsync(); | ||
// Wait for the transaction to finish and execute a retry if necessary. | ||
commitTsFut.get(); | ||
break; | ||
} catch (AbortedException e) { | ||
txn = mgr.resetForRetryAsync(); | ||
} | ||
} | ||
} | ||
|
||
// Calculate the total update count. | ||
ApiFuture<Long> totalUpdateCount = | ||
ApiFutures.transform( | ||
updateCounts, | ||
input -> Arrays.stream(input).sum(), | ||
MoreExecutors.directExecutor()); | ||
System.out.printf("%d records updated.%n", totalUpdateCount.get(30L, TimeUnit.SECONDS)); | ||
executor.shutdown(); | ||
} | ||
} | ||
//[END spanner_postgresql_async_transaction_manager] |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks fixing this syntax issue in old code.