import com.google.rpc.Code; //导入依赖的package包/类
private void preparedTransactionAction(String xid, TransactionAction action) throws SQLException
{
	try
	{
		if (connection.isReadOnly())
		{
			throw new CloudSpannerSQLException(
					"Connection is in read-only mode and cannot be used for prepared transactions",
					Code.FAILED_PRECONDITION);
		}
		else
		{
			action.apply(xid);
		}
	}
	finally
	{
		transactionThread = null;
		readOnlyTransaction = null;
	}
}
 

import com.google.rpc.Code; //导入依赖的package包/类
private CloudSpannerArray(CloudSpannerDataType type, Object[] elements) throws SQLException
{
	this.type = type;
	this.data = java.lang.reflect.Array.newInstance(type.getJavaClass(), elements.length);
	try
	{
		System.arraycopy(elements, 0, this.data, 0, elements.length);
	}
	catch (Exception e)
	{
		throw new CloudSpannerSQLException(
				"Could not copy array elements. Make sure the supplied array only contains elements of class "
						+ type.getJavaClass().getName(),
				Code.UNKNOWN, e);
	}
}
 

import com.google.rpc.Code; //导入依赖的package包/类
@Override
public void commit(Xid xid, boolean onePhase) throws XAException
{
	if (logger.logDebug())
	{
		debug("committing xid = " + xid + (onePhase ? " (one phase) " : " (two phase)"));
	}

	if (xid == null)
	{
		throw new CloudSpannerXAException("xid must not be null", Code.INVALID_ARGUMENT, XAException.XAER_INVAL);
	}

	if (onePhase)
	{
		commitOnePhase(xid);
	}
	else
	{
		commitPrepared(xid);
	}
}
 

import com.google.rpc.Code; //导入依赖的package包/类
@Override
public int executeUpdate(String[] sqlTokens) throws SQLException
{
	if (sqlTokens.length != 1)
		throw new CloudSpannerSQLException(
				"Invalid argument(s) for EXECUTE_DDL_BATCH. Expected \"EXECUTE_DDL_BATCH\"",
				Code.INVALID_ARGUMENT);
	try (CloudSpannerStatement statement = getConnection().createStatement())
	{
		List<String> operations = getConnection().getAutoBatchedDdlOperations();
		for (String sql : operations)
			statement.addBatch(sql);
		statement.executeBatch();
		return operations.size();
	}
	finally
	{
		getConnection().clearAutoBatchedDdlOperations();
	}
}
 

import com.google.rpc.Code; //导入依赖的package包/类
private static Select createSelect(CloudSpannerConnection connection, Delete delete) throws SQLException
{
	TableKeyMetaData table = connection.getTable(CloudSpannerDriver.unquoteIdentifier(delete.getTable().getName()));
	List<String> keyCols = table.getKeyColumns().stream()
			.map(x -> CloudSpannerDriver.quoteIdentifier(delete.getTable().getName()) + "."
					+ CloudSpannerDriver.quoteIdentifier(x))
			.collect(Collectors.toList());
	StringBuilder sql = new StringBuilder();
	sql.append("SELECT ").append(String.join(", ", keyCols));
	sql.append("\nFROM ").append(CloudSpannerDriver.quoteIdentifier(delete.getTable().getName()));
	sql.append("\nWHERE ").append(delete.getWhere().toString());

	try
	{
		return (Select) CCJSqlParserUtil.parse(sql.toString());
	}
	catch (JSQLParserException e)
	{
		throw new CloudSpannerSQLException("Could not parse generated SELECT statement: " + sql,
				Code.INVALID_ARGUMENT);
	}
}
 

import com.google.rpc.Code; //导入依赖的package包/类
public Key.Builder getKeyBuilder() throws SQLException
{
	Key.Builder builder = Key.newBuilder();
	for (String key : table.getKeyColumns())
	{
		Object value = keyValues.get(key);
		if (!generateParameterMetaData && value == null)
		{
			throw new CloudSpannerSQLException(
					"No value supplied for key column " + key
							+ ". All key columns must be specified in the WHERE-clause of a DELETE-statement.",
					Code.INVALID_ARGUMENT);
		}
		builder.appendObject(value);
	}
	return builder;
}
 

import com.google.rpc.Code; //导入依赖的package包/类
private Mutation createUpdateMutation(Update update, boolean generateParameterMetaData) throws SQLException
{
	if (update.getTables().isEmpty())
		throw new CloudSpannerSQLException("No table found in update statement", Code.INVALID_ARGUMENT);
	if (update.getTables().size() > 1)
		throw new CloudSpannerSQLException("Update statements for multiple tables at once are not supported",
				Code.INVALID_ARGUMENT);
	String table = unquoteIdentifier(update.getTables().get(0).getFullyQualifiedName());
	getParameterStore().setTable(table);
	List<Expression> expressions = update.getExpressions();
	WriteBuilder builder = Mutation.newUpdateBuilder(table);
	int index = 0;
	for (Column col : update.getColumns())
	{
		String columnName = unquoteIdentifier(col.getFullyQualifiedName());
		expressions.get(index).accept(new ValueBinderExpressionVisitorAdapter<>(getParameterStore(),
				builder.set(columnName), columnName));
		index++;
	}
	visitUpdateWhereClause(update.getWhere(), builder, generateParameterMetaData);

	return builder.build();
}
 

import com.google.rpc.Code; //导入依赖的package包/类
private void visitDeleteWhereClause(Expression where, DeleteKeyBuilder keyBuilder,
		boolean generateParameterMetaData) throws SQLException
{
	if (where != null)
	{
		DMLWhereClauseVisitor whereClauseVisitor = new DMLWhereClauseVisitor(getParameterStore())
		{

			@Override
			protected void visitExpression(Column col, Expression expression)
			{
				String columnName = unquoteIdentifier(col.getFullyQualifiedName());
				keyBuilder.set(columnName);
				expression.accept(
						new KeyBuilderExpressionVisitorAdapter(getParameterStore(), columnName, keyBuilder));
			}

		};
		where.accept(whereClauseVisitor);
		if (!generateParameterMetaData && !whereClauseVisitor.isValid())
		{
			throw new CloudSpannerSQLException(INVALID_WHERE_CLAUSE_DELETE_MESSAGE, Code.INVALID_ARGUMENT);
		}
	}
}
 

import com.google.rpc.Code; //导入依赖的package包/类
private InsertWorker createInsertWithSelectStatement(Insert insert, boolean forceUpdate) throws SQLException
{
	Select select = insert.getSelect();
	if (select == null)
	{
		throw new CloudSpannerSQLException("Insert statement must contain a select statement",
				Code.INVALID_ARGUMENT);
	}
	boolean isDuplicate = insert.isUseDuplicate();
	InsertWorker.DMLOperation mode;
	if (forceUpdate)
		mode = DMLOperation.Update;
	else if (isDuplicate)
		mode = DMLOperation.OnDuplicateKeyUpdate;
	else
		mode = DMLOperation.Insert;
	return new InsertWorker(getConnection(), select, insert, getConnection().isAllowExtendedMode(), mode);
}
 

import com.google.rpc.Code; //导入依赖的package包/类
void waitForOperations() throws SQLException
{
	boolean finished = false;
	while (!finished)
	{
		finished = true;
		for (DdlOperation op : operations)
		{
			op.operation = op.operation.reload();
			if (!op.operation.isDone())
			{
				finished = false;
				break;
			}
		}
		try
		{
			Thread.sleep(1000l);
		}
		catch (InterruptedException e)
		{
			Thread.currentThread().interrupt();
			throw new CloudSpannerSQLException("Wait for DDL-operations interrupted", Code.CANCELLED, e);
		}
	}
}
 

import com.google.rpc.Code; //导入依赖的package包/类
private static void testCreateTableStatement(String sql) throws SQLException
{
	boolean isDDL = isDDLStatement(sql);
	Assert.assertTrue(isDDL);
	Statement statement = null;
	try
	{
		statement = CCJSqlParserUtil.parse(sql);
	}
	catch (JSQLParserException e)
	{
		throw new CloudSpannerSQLException("Could not parse SQL statement", Code.INVALID_ARGUMENT, e);
	}
	Assert.assertNotNull(statement);
	Assert.assertEquals(CreateTable.class, statement.getClass());
}
 

import com.google.rpc.Code; //导入依赖的package包/类
private Connection createConnection() throws SQLException
{
	try
	{
		Class.forName(CloudSpannerDriver.class.getName());
	}
	catch (ClassNotFoundException e)
	{
		throw new CloudSpannerSQLException("Could not load JDBC driver", Code.UNKNOWN, e);
	}
	StringBuilder url = new StringBuilder("jdbc:cloudspanner://localhost");
	url.append(";Project=").append(projectId);
	url.append(";Instance=").append(instanceId);
	url.append(";Database=").append(DATABASE_ID);
	url.append(";PvtKeyPath=").append(credentialsPath);
	return DriverManager.getConnection(url.toString());
}
 

import com.google.rpc.Code; //导入依赖的package包/类
public void testXA(String projectId, String instanceId, String database, String pvtKeyPath) throws SQLException
{
	log.info("Starting XA tests");
	CloudSpannerXADataSource ds = new CloudSpannerXADataSource();
	ds.setProjectId(projectId);
	ds.setInstanceId(instanceId);
	ds.setDatabase(database);
	ds.setPvtKeyPath(pvtKeyPath);
	ds.setAllowExtendedMode(true);

	try (CloudSpannerXAConnection xaConnection = ds.getXAConnection())
	{
		testXATransaction(xaConnection, CommitMode.TwoPhase);
		testXARollback(xaConnection);
		deleteTestRow(xaConnection);
		testXARecover(xaConnection);
		deleteTestRow(xaConnection);
	}
	catch (Exception e)
	{
		throw new CloudSpannerSQLException("Exception occurred during XA tests", Code.INTERNAL, e);
	}
	log.info("Finished XA tests");
}
 

import com.google.rpc.Code; //导入依赖的package包/类
@Override
public void put(
    Operation operation,
    StreamObserver<com.google.rpc.Status> responseObserver) {
  Instance instance;
  try {
    instance = instances.getFromOperationName(operation.getName());
  } catch (InstanceNotFoundException ex) {
    responseObserver.onError(BuildFarmInstances.toStatusException(ex));
    return;
  }

  boolean ok = instance.putOperation(operation);
  Code code = ok ? Code.OK : Code.UNAVAILABLE;
  responseObserver.onNext(com.google.rpc.Status.newBuilder()
      .setCode(code.getNumber())
      .build());
  responseObserver.onCompleted();
}
 

import com.google.rpc.Code; //导入依赖的package包/类
@Override
public void poll(
    PollOperationRequest request,
    StreamObserver<com.google.rpc.Status> responseObserver) {
  Instance instance;
  try {
    instance = instances.getFromOperationName(
        request.getOperationName());
  } catch (InstanceNotFoundException ex) {
    responseObserver.onError(BuildFarmInstances.toStatusException(ex));
    return;
  }

  boolean ok = instance.pollOperation(
      request.getOperationName(),
      request.getStage());
  Code code = ok ? Code.OK : Code.UNAVAILABLE;
  responseObserver.onNext(com.google.rpc.Status.newBuilder()
      .setCode(code.getNumber())
      .build());
  responseObserver.onCompleted();
}
 

import com.google.rpc.Code; //导入依赖的package包/类
/**
 * Translates a Cloud Datastore gql query string to {@link Query}.
 *
 * <p>Currently, the only way to translate a gql query string to a Query is to run the query
 * against Cloud Datastore and extract the {@code Query} from the response. To prevent reading
 * any data, we set the {@code LIMIT} to 0 but if the gql query already has a limit set, we
 * catch the exception with {@code INVALID_ARGUMENT} error code and retry the translation
 * without the zero limit.
 *
 * <p>Note: This may result in reading actual data from Cloud Datastore but the service has a
 * cap on the number of entities returned for a single rpc request, so this should not be a
 * problem in practice.
 */
@VisibleForTesting
static Query translateGqlQueryWithLimitCheck(String gql, Datastore datastore,
    String namespace) throws DatastoreException {
  String gqlQueryWithZeroLimit = gql + " LIMIT 0";
  try {
    Query translatedQuery = translateGqlQuery(gqlQueryWithZeroLimit, datastore, namespace);
    // Clear the limit that we set.
    return translatedQuery.toBuilder().clearLimit().build();
  } catch (DatastoreException e) {
    // Note: There is no specific error code or message to detect if the query already has a
    // limit, so we just check for INVALID_ARGUMENT and assume that that the query might have
    // a limit already set.
    if (e.getCode() == Code.INVALID_ARGUMENT) {
      LOG.warn("Failed to translate Gql query '{}': {}", gqlQueryWithZeroLimit, e.getMessage());
      LOG.warn("User query might have a limit already set, so trying without zero limit");
      // Retry without the zero limit.
      return translateGqlQuery(gql, datastore, namespace);
    } else {
      throw e;
    }
  }
}
 

import com.google.rpc.Code; //导入依赖的package包/类
/** Tests {@link DatastoreWriterFn} with a failed request which is retried. */
@Test
public void testDatatoreWriterFnRetriesErrors() throws Exception {
  List<Mutation> mutations = new ArrayList<>();
  int numRpcs = 2;
  for (int i = 0; i < DATASTORE_BATCH_UPDATE_ENTITIES_START * numRpcs; ++i) {
    mutations.add(
        makeUpsert(Entity.newBuilder().setKey(makeKey("key" + i, i + 1)).build()).build());
  }

  CommitResponse successfulCommit = CommitResponse.getDefaultInstance();
  when(mockDatastore.commit(any(CommitRequest.class))).thenReturn(successfulCommit)
    .thenThrow(
        new DatastoreException("commit", Code.DEADLINE_EXCEEDED, "", null))
    .thenReturn(successfulCommit);

  DatastoreWriterFn datastoreWriter = new DatastoreWriterFn(StaticValueProvider.of(PROJECT_ID),
      null, mockDatastoreFactory, new FakeWriteBatcher());
  DoFnTester<Mutation, Void> doFnTester = DoFnTester.of(datastoreWriter);
  doFnTester.setCloningBehavior(CloningBehavior.DO_NOT_CLONE);
  doFnTester.processBundle(mutations);
}
 

import com.google.rpc.Code; //导入依赖的package包/类
/** Tests that {@link ReadFn} retries after an error. */
@Test
public void testReadFnRetriesErrors() throws Exception {
  // An empty query to read entities.
  Query query = Query.newBuilder().setLimit(
      Int32Value.newBuilder().setValue(1)).build();

  // Use mockResponseForQuery to generate results.
  when(mockDatastore.runQuery(any(RunQueryRequest.class)))
      .thenThrow(
          new DatastoreException("RunQuery", Code.DEADLINE_EXCEEDED, "", null))
      .thenAnswer(new Answer<RunQueryResponse>() {
        @Override
        public RunQueryResponse answer(InvocationOnMock invocationOnMock) throws Throwable {
          Query q = ((RunQueryRequest) invocationOnMock.getArguments()[0]).getQuery();
          return mockResponseForQuery(q);
        }
      });

  ReadFn readFn = new ReadFn(V_1_OPTIONS, mockDatastoreFactory);
  DoFnTester<Query, Entity> doFnTester = DoFnTester.of(readFn);
  doFnTester.setCloningBehavior(CloningBehavior.DO_NOT_CLONE);
  List<Entity> entities = doFnTester.processBundle(query);
}
 

import com.google.rpc.Code; //导入依赖的package包/类
@Test
public void testTranslateGqlQueryWithLimit() throws Exception {
  String gql = "SELECT * from DummyKind LIMIT 10";
  String gqlWithZeroLimit = gql + " LIMIT 0";
  GqlQuery gqlQuery = GqlQuery.newBuilder().setQueryString(gql).setAllowLiterals(true).build();
  GqlQuery gqlQueryWithZeroLimit = GqlQuery.newBuilder().setQueryString(gqlWithZeroLimit)
      .setAllowLiterals(true).build();
  RunQueryRequest gqlRequest = makeRequest(gqlQuery, V_1_OPTIONS.getNamespace());
  RunQueryRequest gqlRequestWithZeroLimit = makeRequest(gqlQueryWithZeroLimit,
      V_1_OPTIONS.getNamespace());
  when(mockDatastore.runQuery(gqlRequestWithZeroLimit))
      .thenThrow(new DatastoreException("runQuery", Code.INVALID_ARGUMENT, "invalid query",
          // dummy
          new RuntimeException()));
  when(mockDatastore.runQuery(gqlRequest))
      .thenReturn(RunQueryResponse.newBuilder().setQuery(QUERY).build());
  assertEquals(translateGqlQueryWithLimitCheck(gql, mockDatastore, V_1_OPTIONS.getNamespace()),
      QUERY);
  verify(mockDatastore, times(1)).runQuery(gqlRequest);
  verify(mockDatastore, times(1)).runQuery(gqlRequestWithZeroLimit);
}
 

import com.google.rpc.Code; //导入依赖的package包/类
@Override
public void batchUpdateBlobs(
    BatchUpdateBlobsRequest request, StreamObserver<BatchUpdateBlobsResponse> responseObserver) {
  BatchUpdateBlobsResponse.Builder batchResponse = BatchUpdateBlobsResponse.newBuilder();
  for (UpdateBlobRequest r : request.getRequestsList()) {
    BatchUpdateBlobsResponse.Response.Builder resp = batchResponse.addResponsesBuilder();
    try {
      Digest digest = cache.uploadBlob(r.getData().toByteArray());
      if (!r.getContentDigest().equals(digest)) {
        String err =
            "Upload digest " + r.getContentDigest() + " did not match data digest: " + digest;
        resp.setStatus(StatusUtils.invalidArgumentStatus("content_digest", err));
        continue;
      }
      resp.getStatusBuilder().setCode(Code.OK.getNumber());
    } catch (Exception e) {
      resp.setStatus(StatusUtils.internalErrorStatus(e));
    }
  }
  responseObserver.onNext(batchResponse.build());
  responseObserver.onCompleted();
}