amaliujia commented on a change in pull request #13319:
URL: https://github.com/apache/beam/pull/13319#discussion_r529004528
##########
File path:
website/www/site/content/en/documentation/dsls/sql/extensions/create-external-table.md
##########
@@ -204,6 +205,131 @@ TYPE bigquery
LOCATION 'testing-integration:apache.users'
```
+## Cloud Bigtable
+
+### Syntax
+
+```
+CREATE EXTERNAL TABLE [ IF NOT EXISTS ] tableName (
+ key VARCHAR NOT NULL,
+ family ROW<qualifier cells [, qualifier cells ]* >
+ [, family ROW< qualifier cells [, qualifier cells ]* > ]*
+)
+TYPE bigtable
+LOCATION
'googleapis.com/bigtable/projects/[PROJECT_ID]/instances/[INSTANCE_ID]/tables/[TABLE]'
+```
+
+* `key`: key of the Bigtable row
+* `family`: name of the column family
+* `qualifier`: the column qualifier
+* `cells`: Either of each value:
+ * `TYPE`
+ * `ARRAY<SIMPLE_TYPE>`
+* `LOCATION`:
+ * `PROJECT_ID`: ID of the Google Cloud Project.
+ * `INSTANCE_ID`: Bigtable instance ID.
+ * `TABLE`: Bigtable Table ID.
+* `TYPE`: `SIMPLE_TYPE` or `CELL_ROW`
+* `CELL_ROW`: `ROW<val SIMPLE_TYPE [, timestampMicros BIGINT [NOT NULL]] [,
labels ARRAY<VARCHAR> [NOT NULL]]`
+* `SIMPLE_TYPE`: on of the following:
+ * `BINARY`
+ * `VARCHAR`
+ * `BIGINT`
+ * `DOUBLE`
+ * `BOOLEAN`
Review comment:
So what is the reason that single type only supports these 5 types (i.e.
not include other SQL types like TIMESTAMP)?
##########
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/bigtable/BigtableTable.java
##########
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.extensions.sql.meta.provider.bigtable;
+
+import static java.util.stream.Collectors.toSet;
+import static org.apache.beam.sdk.io.gcp.bigtable.RowUtils.COLUMNS_MAPPING;
+import static org.apache.beam.sdk.io.gcp.bigtable.RowUtils.KEY;
+import static
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayList;
+
+import com.alibaba.fastjson.JSONObject;
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.extensions.sql.impl.BeamTableStatistics;
+import org.apache.beam.sdk.extensions.sql.meta.SchemaBaseBeamTable;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+import org.apache.beam.sdk.extensions.sql.meta.provider.InvalidTableException;
+import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
+import org.apache.beam.sdk.io.gcp.bigtable.BigtableRowToBeamRow;
+import org.apache.beam.sdk.io.gcp.bigtable.BigtableRowToBeamRowFlat;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Splitter;
+
+@Experimental
+public class BigtableTable extends SchemaBaseBeamTable implements Serializable
{
+ // Should match:
+ //
googleapis.com/bigtable/projects/projectId/instances/instanceId/tables/tableId"
+ private static final Pattern locationPattern =
+ Pattern.compile(
+
"(?<host>.+)/bigtable/projects/(?<projectId>.+)/instances/(?<instanceId>.+)/tables/(?<tableId>.+)");
+
+ private final String projectId;
+ private final String instanceId;
+ private final String tableId;
+ private String emulatorHost = "";
+
+ private boolean useFlatSchema = false;
+
+ private Map<String, List<String>> columnsMapping = new HashMap<>();
+
+ BigtableTable(Table table) {
+ super(table.getSchema());
+ validateSchema(schema);
+
+ String location = table.getLocation();
+ if (location == null) {
+ throw new IllegalStateException("LOCATION is required");
+ }
+ Matcher matcher = locationPattern.matcher(location);
+ validateMatcher(matcher, location);
+
+ this.projectId = getMatcherValue(matcher, "projectId");
+ this.instanceId = getMatcherValue(matcher, "instanceId");
+ this.tableId = getMatcherValue(matcher, "tableId");
+ String host = getMatcherValue(matcher, "host"); // googleapis.com or
localhost:<PORT>
+ if (!"googleapis.com".equals(host)) {
+ this.emulatorHost = host;
+ }
+
+ JSONObject properties = table.getProperties();
+ if (properties.containsKey(COLUMNS_MAPPING)) {
+ columnsMapping =
parseColumnsMapping(properties.getString(COLUMNS_MAPPING));
+ validateColumnsMapping(columnsMapping, schema);
+ useFlatSchema = true;
+ }
+ }
+
+ @Override
+ public PCollection<Row> buildIOReader(PBegin begin) {
+ BigtableIO.Read readTransform =
+
BigtableIO.read().withProjectId(projectId).withInstanceId(instanceId).withTableId(tableId);
+ if (!emulatorHost.isEmpty()) {
+ readTransform = readTransform.withEmulator(emulatorHost);
+ }
+ return readTransform
+ .expand(begin)
+ .apply(
+ "BigtableRowToBeamRow",
+ useFlatSchema
+ ? new BigtableRowToBeamRowFlat(schema, columnsMapping)
+ : new BigtableRowToBeamRow(schema))
+ .setRowSchema(schema);
+ }
+
+ @Override
+ public POutput buildIOWriter(PCollection<Row> input) {
+ throw new UnsupportedOperationException("Write to Cloud Bigtable is not
yet supported");
+ }
+
+ @Override
+ public PCollection.IsBounded isBounded() {
+ return PCollection.IsBounded.BOUNDED;
+ }
+
+ @Override
+ public BeamTableStatistics getTableStatistics(PipelineOptions options) {
+ return BeamTableStatistics.BOUNDED_UNKNOWN;
+ }
+
+ private static Map<String, List<String>> parseColumnsMapping(String
commaSeparatedMapping) {
Review comment:
for this functions and other functions below, as they are already
private functions, maybe should remove the `static`?
##########
File path:
sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/bigtable/BigtableTableCreationFailuresTest.java
##########
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.extensions.sql.meta.provider.bigtable;
+
+import static
org.apache.beam.sdk.io.gcp.testing.BigtableTestUtils.checkMessage;
+import static org.junit.Assert.assertThrows;
+
+import org.apache.beam.sdk.extensions.sql.BeamSqlCli;
+import org.apache.beam.sdk.extensions.sql.impl.ParseException;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider;
+import org.apache.beam.sdk.extensions.sql.meta.store.InMemoryMetaStore;
+import org.junit.Before;
+import org.junit.Test;
+
+@SuppressWarnings({
+ "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
+})
+public class BigtableTableCreationFailuresTest {
+
+ private final InMemoryMetaStore metaStore = new InMemoryMetaStore();
+ private final TableProvider tableProvider = new BigtableTableProvider();
+ private BeamSqlCli cli;
+
+ @Before
+ public void setUp() {
+ metaStore.registerProvider(tableProvider);
+ cli = new BeamSqlCli().metaStore(metaStore);
+ }
+
+ @Test
+ public void testCreateWithoutTypeFails() {
+ String createTable = "CREATE EXTERNAL TABLE failure(something VARCHAR)";
+ ParseException e = assertThrows(ParseException.class, () ->
cli.execute(createTable));
+ checkMessage(e.getMessage(), "Unable to parse query");
+ }
+
+ @Test
+ public void testCreateWithoutLocationFails() {
+ String createTable =
+ "CREATE EXTERNAL TABLE fail(key VARCHAR, something VARCHAR) \n" +
"TYPE bigtable \n";
+ cli.execute(createTable);
+ Table table = metaStore.getTables().get("fail");
+
+ IllegalStateException e =
+ assertThrows(IllegalStateException.class, () ->
tableProvider.buildBeamSqlTable(table));
+ checkMessage(e.getMessage(), "LOCATION");
+ }
+
+ @Test
+ public void testCreateWithoutKeyFails() {
+ String createTable =
+ "CREATE EXTERNAL TABLE fail(something VARCHAR) \n"
+ + "TYPE bigtable \n"
+ + "LOCATION '"
+ + location()
+ + "'";
+ cli.execute(createTable);
+ Table table = metaStore.getTables().get("fail");
+ IllegalStateException e =
+ assertThrows(IllegalStateException.class, () ->
tableProvider.buildBeamSqlTable(table));
+ checkMessage(e.getMessage(), "Schema has to contain 'key' field");
+ }
+
+ @Test
+ public void testCreateWrongKeyTypeFails() {
+ String createTable =
+ "CREATE EXTERNAL TABLE fail(key FLOAT) \n"
+ + "TYPE bigtable \n"
+ + "LOCATION '"
+ + location()
+ + "'";
+ cli.execute(createTable);
+ Table table = metaStore.getTables().get("fail");
+ IllegalArgumentException e =
+ assertThrows(IllegalArgumentException.class, () ->
tableProvider.buildBeamSqlTable(table));
+ checkMessage(e.getMessage(), "key field type should be STRING but was
FLOAT");
+ }
+
+ @Test
+ public void testCreatePropertiesDontMatchSchema() {
+ String createTable =
+ "CREATE EXTERNAL TABLE fail(key VARCHAR, q BIGINT, qq BINARY) \n"
+ + "TYPE bigtable \n"
+ + "LOCATION '"
+ + location()
+ + "' \n"
+ + "TBLPROPERTIES '{\"columnsMapping\": \"f:b,f:c\"}'";
+ cli.execute(createTable);
+ Table table = metaStore.getTables().get("fail");
+ IllegalStateException e =
+ assertThrows(IllegalStateException.class, () ->
tableProvider.buildBeamSqlTable(table));
+ checkMessage(e.getMessage(), "does not fit to schema field names");
+ }
+
+ @Test
+ public void testCreatePropertiesCountNotEqualSchemaFields() {
+ String createTable =
+ "CREATE EXTERNAL TABLE fail(key VARCHAR, q BIGINT, qq BINARY) \n"
+ + "TYPE bigtable \n"
+ + "LOCATION '"
+ + location()
+ + "' \n"
+ + "TBLPROPERTIES '{\"columnsMapping\": \"f:q\"}'";
+ cli.execute(createTable);
+ Table table = metaStore.getTables().get("fail");
+ IllegalStateException e =
+ assertThrows(IllegalStateException.class, () ->
tableProvider.buildBeamSqlTable(table));
+ checkMessage(e.getMessage(), "Schema fields count: '2' does not fit
columnsMapping count: '1'");
+ }
+
+ private static String location() {
+ return
"googleapis.com/bigtable/projects/fakeProject/instances/fakeInstance/tables/beamTable";
+ }
+}
Review comment:
Add a test to hit
```
private static void validateMatcher(Matcher matcher, String location) {
if (!matcher.matches()) {
throw new InvalidTableException(
"Bigtable location must be in the following format:"
+ "
'googleapis.com/bigtable/projects/projectId/instances/instanceId/tables/tableId'"
+ " but was: "
+ location);
}
}
```
?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
