diff --git a/.test-infra/tools/stale_bq_datasets_cleaner.sh b/.test-infra/tools/stale_bq_datasets_cleaner.sh index 652deaddcc56..36f772397419 100755 --- a/.test-infra/tools/stale_bq_datasets_cleaner.sh +++ b/.test-infra/tools/stale_bq_datasets_cleaner.sh @@ -24,7 +24,7 @@ PROJECT=apache-beam-testing MAX_RESULT=1500 BQ_DATASETS=`bq --project_id=$PROJECT ls --max_results=$MAX_RESULT | tail -n $MAX_RESULT | sed s/^[[:space:]]*/${PROJECT}:/` -CLEANUP_DATASET_TEMPLATES=(beam_bigquery_samples_ beam_temp_dataset_ FHIR_store_ bq_query_schema_update_options_16 bq_query_to_table_16 '\:(bq_read_all_|combine_per_key_examples|filter_examples|hourly_team_score_|leader_board_|leaderboard_|game_stats_|python_|temp_dataset)[a-z_]*[0-9a-f]{12,}$') +CLEANUP_DATASET_TEMPLATES=( beam_temp_dataset_ file_loads_streaming_it_ FHIR_store_ io_bigquery_lt_ managed-iceberg-biglake-its.biglake_test_catalog_ 'managed_iceberg_bqms.*_tests?_' '_(python|java)_validations_' '\:(bq_read_all_|combine_per_key_examples|filter_examples|hourly_team_score_|leader_board_|leaderboard_|game_stats_|python_|temp_dataset)[a-z_]*[0-9a-f]{12,}$') # A grace period of 5 days GRACE_PERIOD=$((`date +%s` - 24 * 3600 * 5)) diff --git a/examples/java/src/test/java/org/apache/beam/examples/complete/TrafficMaxLaneFlowIT.java b/examples/java/src/test/java/org/apache/beam/examples/complete/TrafficMaxLaneFlowIT.java index 1832d00aec1d..402ef13e944b 100644 --- a/examples/java/src/test/java/org/apache/beam/examples/complete/TrafficMaxLaneFlowIT.java +++ b/examples/java/src/test/java/org/apache/beam/examples/complete/TrafficMaxLaneFlowIT.java @@ -17,6 +17,7 @@ */ package org.apache.beam.examples.complete; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; import static org.junit.Assert.assertEquals; import com.google.api.client.util.BackOff; @@ -44,7 +45,7 @@ public class TrafficMaxLaneFlowIT { private TrafficMaxLaneFlowOptions options; private final String timestamp = Long.toString(System.currentTimeMillis()); - private final String outputDatasetId = "traffic_max_lane_flow_" + timestamp; + private final String outputDatasetId = TEMP_DATASET_PREFIX + "traffic_max_lane_flow_" + timestamp; private final String outputTable = "traffic_max_lane_flow_table"; private String projectId; private BigqueryClient bqClient; diff --git a/examples/java/src/test/java/org/apache/beam/examples/cookbook/MaxPerKeyExamplesIT.java b/examples/java/src/test/java/org/apache/beam/examples/cookbook/MaxPerKeyExamplesIT.java index 16ea27d91323..53eb3413d57f 100644 --- a/examples/java/src/test/java/org/apache/beam/examples/cookbook/MaxPerKeyExamplesIT.java +++ b/examples/java/src/test/java/org/apache/beam/examples/cookbook/MaxPerKeyExamplesIT.java @@ -17,6 +17,7 @@ */ package org.apache.beam.examples.cookbook; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; import static org.junit.Assert.assertEquals; import com.google.api.client.util.BackOff; @@ -42,7 +43,7 @@ public class MaxPerKeyExamplesIT { private MaxPerKeyExamplesIT.MaxPerKeyExamplesOptions options; private final String timestamp = Long.toString(System.currentTimeMillis()); - private final String outputDatasetId = "max_per_key_examples" + timestamp; + private final String outputDatasetId = TEMP_DATASET_PREFIX + "maxperkey_" + timestamp; private final String outputTable = "max_per_key_examples_table"; private final Long defaultExpiration = 1000L * 60 * 60; private String projectId; diff --git a/examples/java/src/test/java/org/apache/beam/examples/snippets/transforms/io/gcp/bigquery/BigQuerySamplesIT.java b/examples/java/src/test/java/org/apache/beam/examples/snippets/transforms/io/gcp/bigquery/BigQuerySamplesIT.java index cdaffa901bc8..a14342362778 100644 --- a/examples/java/src/test/java/org/apache/beam/examples/snippets/transforms/io/gcp/bigquery/BigQuerySamplesIT.java +++ b/examples/java/src/test/java/org/apache/beam/examples/snippets/transforms/io/gcp/bigquery/BigQuerySamplesIT.java @@ -17,6 +17,7 @@ */ package org.apache.beam.examples.snippets.transforms.io.gcp.bigquery; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; import static org.junit.Assert.assertEquals; import com.google.api.services.bigquery.model.TableRow; @@ -76,7 +77,11 @@ public class BigQuerySamplesIT { private static final BigQuery BIGQUERY = BigQueryOptions.newBuilder().setProjectId(PROJECT).build().getService(); private static final String DATASET = - "beam_bigquery_samples_" + System.currentTimeMillis() + "_" + new SecureRandom().nextInt(32); + TEMP_DATASET_PREFIX + + "samples_" + + System.currentTimeMillis() + + "_" + + new SecureRandom().nextInt(32); @Rule public final transient TestPipeline writePipeline = TestPipeline.create(); @Rule public final transient TestPipeline readTablePipeline = TestPipeline.create(); diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryStreamingLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryStreamingLT.java index 1b7257a8cde5..7715f782b548 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryStreamingLT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryStreamingLT.java @@ -19,6 +19,7 @@ import static org.apache.beam.sdk.io.gcp.bigquery.BigQueryUtils.toTableReference; import static org.apache.beam.sdk.io.gcp.bigquery.BigQueryUtils.toTableSpec; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotEquals; @@ -95,7 +96,7 @@ public class BigQueryStreamingLT extends IOLoadTestBase { private static final BigqueryClient BQ_CLIENT = new BigqueryClient("BigQueryStreamingLT"); private static final String BIG_QUERY_DATASET_ID = - "storage_api_sink_load_test_" + System.nanoTime(); + TEMP_DATASET_PREFIX + "sink_load_test_" + System.nanoTime(); private TestConfiguration config; private Integer crashIntervalSeconds; diff --git a/sdks/java/extensions/sql/iceberg/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergReadWriteIT.java b/sdks/java/extensions/sql/iceberg/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergReadWriteIT.java index 3a791c6fe88c..5b47fca79ce2 100644 --- a/sdks/java/extensions/sql/iceberg/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergReadWriteIT.java +++ b/sdks/java/extensions/sql/iceberg/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergReadWriteIT.java @@ -19,6 +19,7 @@ import static java.lang.String.format; import static java.util.Arrays.asList; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; import static org.apache.beam.sdk.schemas.Schema.FieldType.BOOLEAN; import static org.apache.beam.sdk.schemas.Schema.FieldType.DOUBLE; import static org.apache.beam.sdk.schemas.Schema.FieldType.FLOAT; @@ -101,7 +102,7 @@ public class IcebergReadWriteIT { private static final BigqueryClient BQ_CLIENT = new BigqueryClient("IcebergReadWriteIT"); private static final String BQMS_CATALOG = "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog"; - static final String DATASET = "iceberg_sql_tests_" + System.nanoTime(); + static final String DATASET = TEMP_DATASET_PREFIX + "iceberg_sql_" + System.nanoTime(); static String warehouse; protected static final GcpOptions OPTIONS = TestPipeline.testingPipelineOptions().as(GcpOptions.class); diff --git a/sdks/java/extensions/sql/iceberg/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/PubsubToIcebergIT.java b/sdks/java/extensions/sql/iceberg/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/PubsubToIcebergIT.java index 8b250af2754a..5d9475a8c4b8 100644 --- a/sdks/java/extensions/sql/iceberg/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/PubsubToIcebergIT.java +++ b/sdks/java/extensions/sql/iceberg/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/PubsubToIcebergIT.java @@ -19,6 +19,7 @@ import static java.lang.String.format; import static java.nio.charset.StandardCharsets.UTF_8; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; import static org.apache.beam.sdk.schemas.Schema.FieldType.INT64; import static org.apache.beam.sdk.schemas.Schema.FieldType.STRING; import static org.junit.Assert.assertEquals; @@ -76,7 +77,7 @@ public class PubsubToIcebergIT implements Serializable { private static final BigqueryClient BQ_CLIENT = new BigqueryClient("PubsubToIcebergIT"); private static final String BQMS_CATALOG = "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog"; - static final String DATASET = "sql_pubsub_to_iceberg_it_" + System.nanoTime(); + static final String DATASET = TEMP_DATASET_PREFIX + "sql_pubsub_to_iceberg_" + System.nanoTime(); static String warehouse; private static Catalog icebergCatalog; protected static final GcpOptions OPTIONS = diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/BigqueryClient.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/BigqueryClient.java index 151d94f0349e..a83c49404912 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/BigqueryClient.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/BigqueryClient.java @@ -43,8 +43,6 @@ import com.google.api.services.bigquery.model.TableDataInsertAllRequest.Rows; import com.google.api.services.bigquery.model.TableDataInsertAllResponse; import com.google.api.services.bigquery.model.TableFieldSchema; -import com.google.api.services.bigquery.model.TableList; -import com.google.api.services.bigquery.model.TableList.Tables; import com.google.api.services.bigquery.model.TableReference; import com.google.api.services.bigquery.model.TableRow; import com.google.api.services.bigquery.model.TableSchema; @@ -482,16 +480,7 @@ public void deleteTable(String projectId, String datasetId, String tableName) { public void deleteDataset(String projectId, String datasetId) { try { - TableList tables = bqClient.tables().list(projectId, datasetId).execute(); - for (Tables table : tables.getTables()) { - this.deleteTable(projectId, datasetId, table.getTableReference().getTableId()); - } - } catch (Exception e) { - LOG.debug("Exceptions caught when listing all tables", e); - } - - try { - bqClient.datasets().delete(projectId, datasetId).execute(); + bqClient.datasets().delete(projectId, datasetId).setDeleteContents(true).execute(); LOG.info("Successfully deleted dataset: {}", datasetId); } catch (Exception e) { LOG.debug("Exceptions caught when deleting dataset", e); diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/BigtableUtils.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/BigtableUtils.java index 0d1a5e05b172..3cf45cfc0015 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/BigtableUtils.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/BigtableUtils.java @@ -23,6 +23,8 @@ public class BigtableUtils { + public static final String TEMP_DATASET_PREFIX = "beam_temp_dataset_"; + public static ByteString byteString(byte[] bytes) { return ByteString.copyFrom(bytes); } diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryKmsKeyIT.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryKmsKeyIT.java index edf93dba3ae4..b85b6d1e54c6 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryKmsKeyIT.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryKmsKeyIT.java @@ -17,6 +17,7 @@ */ package org.apache.beam.sdk.io.gcp.bigquery; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -56,7 +57,11 @@ public class BigQueryKmsKeyIT { private static final BigqueryClient BQ_CLIENT = new BigqueryClient("BigQueryKmsKeyIT"); private static final String BIG_QUERY_DATASET_ID = - "bq_query_to_table_" + System.currentTimeMillis() + "_" + new SecureRandom().nextInt(32); + TEMP_DATASET_PREFIX + + "query_to_table_" + + System.currentTimeMillis() + + "_" + + new SecureRandom().nextInt(32); private static final TableSchema OUTPUT_SCHEMA = new TableSchema() .setFields(ImmutableList.of(new TableFieldSchema().setName("fruit").setType("STRING"))); diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySchemaUpdateOptionsIT.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySchemaUpdateOptionsIT.java index c5e954a46380..9ac2cc4bc088 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySchemaUpdateOptionsIT.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySchemaUpdateOptionsIT.java @@ -17,6 +17,7 @@ */ package org.apache.beam.sdk.io.gcp.bigquery; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; import static org.junit.Assert.assertEquals; import com.google.api.services.bigquery.model.QueryResponse; @@ -68,7 +69,8 @@ public class BigQuerySchemaUpdateOptionsIT { new BigqueryClient("BigQuerySchemaUpdateOptionsIT"); private static final String BIG_QUERY_DATASET_ID = - "bq_query_schema_update_options_" + TEMP_DATASET_PREFIX + + "schema_update_options_" + System.currentTimeMillis() + "_" + new SecureRandom().nextInt(32); diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTimePartitioningClusteringIT.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTimePartitioningClusteringIT.java index 008cc02beee3..848d4c2ea6ec 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTimePartitioningClusteringIT.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTimePartitioningClusteringIT.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.io.gcp.bigquery; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; + import com.google.api.services.bigquery.Bigquery; import com.google.api.services.bigquery.model.Clustering; import com.google.api.services.bigquery.model.Table; @@ -59,7 +61,8 @@ public class BigQueryTimePartitioningClusteringIT { private static final BigqueryClient BQ_CLIENT = new BigqueryClient("BigQueryTimePartitioningClusteringIT"); private static final String DATASET_NAME = - "BigQueryTimePartitioningIT_" + TEMP_DATASET_PREFIX + + "time_partitioning_" + System.currentTimeMillis() + "_" + new SecureRandom().nextInt(32); diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTimestampPicosIT.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTimestampPicosIT.java index 07b6adf46bcd..a5303240052d 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTimestampPicosIT.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTimestampPicosIT.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.io.gcp.bigquery; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; + import com.google.api.services.bigquery.model.TableFieldSchema; import com.google.api.services.bigquery.model.TableRow; import com.google.api.services.bigquery.model.TableSchema; @@ -56,7 +58,11 @@ public class BigQueryTimestampPicosIT { private static String project; private static final String DATASET_ID = - "bq_ts_picos_" + System.currentTimeMillis() + "_" + new SecureRandom().nextInt(32); + TEMP_DATASET_PREFIX + + "ts_picos_" + + System.currentTimeMillis() + + "_" + + new SecureRandom().nextInt(32); private static final BigqueryClient BQ_CLIENT = new BigqueryClient("BigQueryTimestampPicosIT"); private static TestBigQueryOptions bqOptions; private static String nestedTableSpec; diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryToTableIT.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryToTableIT.java index 31bb627e8451..407f00142cd1 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryToTableIT.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryToTableIT.java @@ -17,6 +17,7 @@ */ package org.apache.beam.sdk.io.gcp.bigquery; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; import static org.junit.Assert.assertEquals; import com.google.api.client.util.BackOff; @@ -72,7 +73,11 @@ public class BigQueryToTableIT { private static final BigqueryClient BQ_CLIENT = new BigqueryClient("BigQueryToTableIT"); private static final String BIG_QUERY_DATASET_ID = - "bq_query_to_table_" + System.currentTimeMillis() + "_" + new SecureRandom().nextInt(32); + TEMP_DATASET_PREFIX + + "query_to_table_" + + System.currentTimeMillis() + + "_" + + new SecureRandom().nextInt(32); private static final TableSchema LEGACY_QUERY_TABLE_SCHEMA = new TableSchema() diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiDataTriggeredSchemaUpdateIT.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiDataTriggeredSchemaUpdateIT.java index d90dedaf06f6..25cc017a1ca7 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiDataTriggeredSchemaUpdateIT.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiDataTriggeredSchemaUpdateIT.java @@ -17,6 +17,7 @@ */ package org.apache.beam.sdk.io.gcp.bigquery; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects.firstNonNull; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -82,7 +83,7 @@ public class StorageApiDataTriggeredSchemaUpdateIT { private static final String PROJECT = TestPipeline.testingPipelineOptions().as(GcpOptions.class).getProject(); private static final String BIG_QUERY_DATASET_ID = - "storage_api_data_triggered_schema_update_" + System.nanoTime(); + TEMP_DATASET_PREFIX + "triggered_schema_update_" + System.nanoTime(); private static String bigQueryLocation; diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkCreateIfNeededIT.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkCreateIfNeededIT.java index 858921e19ced..70eae4095982 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkCreateIfNeededIT.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkCreateIfNeededIT.java @@ -19,6 +19,7 @@ import static org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.CONNECTION_ID; import static org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.STORAGE_URI; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; import static org.hamcrest.MatcherAssert.assertThat; import static org.junit.Assert.assertEquals; @@ -71,7 +72,7 @@ public static Iterable data() { private static final String PROJECT = TestPipeline.testingPipelineOptions().as(GcpOptions.class).getProject(); private static final String BIG_QUERY_DATASET_ID = - "storage_api_sink_create_tables_" + System.nanoTime(); + TEMP_DATASET_PREFIX + "storageapi_" + System.nanoTime(); private static final String TEST_CONNECTION_ID = "projects/apache-beam-testing/locations/us/connections/apache-beam-testing-storageapi-biglake-nodelete"; private static final String TEST_STORAGE_URI = diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkFailedRowsIT.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkFailedRowsIT.java index f721f57147e3..5124e46847d9 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkFailedRowsIT.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkFailedRowsIT.java @@ -17,6 +17,7 @@ */ package org.apache.beam.sdk.io.gcp.bigquery; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; import static org.hamcrest.MatcherAssert.assertThat; import com.google.api.services.bigquery.model.Table; @@ -74,7 +75,7 @@ public static Iterable data() { private static final String PROJECT = TestPipeline.testingPipelineOptions().as(GcpOptions.class).getProject(); private static final String BIG_QUERY_DATASET_ID = - "storage_api_sink_failed_rows" + System.nanoTime(); + TEMP_DATASET_PREFIX + "failed_rows" + System.nanoTime(); private static final List FIELDS = ImmutableList.builder() diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkSchemaUpdateWithInputSchemaIT.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkSchemaUpdateWithInputSchemaIT.java index cd29d9fb65a1..b6ef1ce7290e 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkSchemaUpdateWithInputSchemaIT.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkSchemaUpdateWithInputSchemaIT.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.io.gcp.bigquery; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; + import java.io.IOException; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; import org.junit.AfterClass; @@ -27,7 +29,7 @@ @RunWith(Parameterized.class) public class StorageApiSinkSchemaUpdateWithInputSchemaIT extends StorageApiSinkSchemaUpdateITBase { private static final String BIG_QUERY_DATASET_ID = - "storage_api_sink_schema_change_with_input_" + System.nanoTime(); + TEMP_DATASET_PREFIX + "sink_schema_change_with_input_" + System.nanoTime(); @Parameterized.Parameters(name = "changeTableSchema={0}") public static Iterable data() { diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkSchemaUpdateWithoutInputSchemaIT.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkSchemaUpdateWithoutInputSchemaIT.java index 9b02e0be396b..ed9d13ee4fde 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkSchemaUpdateWithoutInputSchemaIT.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkSchemaUpdateWithoutInputSchemaIT.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.io.gcp.bigquery; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; + import java.io.IOException; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; import org.junit.AfterClass; @@ -28,7 +30,7 @@ public class StorageApiSinkSchemaUpdateWithoutInputSchemaIT extends StorageApiSinkSchemaUpdateITBase { private static final String BIG_QUERY_DATASET_ID = - "storage_api_sink_schema_change_without_input_" + System.nanoTime(); + TEMP_DATASET_PREFIX + "sink_schema_change_without_input_" + System.nanoTime(); @Parameterized.Parameters(name = "changeTableSchema={0}") public static Iterable data() { diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryManagedIT.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryManagedIT.java index 3106063b2d9a..44fd7ecfe7f6 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryManagedIT.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryManagedIT.java @@ -17,6 +17,7 @@ */ package org.apache.beam.sdk.io.gcp.bigquery.providers; +import static org.apache.beam.sdk.io.gcp.testing.BigtableUtils.TEMP_DATASET_PREFIX; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.containsInAnyOrder; @@ -83,7 +84,8 @@ public class BigQueryManagedIT { private static final String PROJECT = TestPipeline.testingPipelineOptions().as(GcpOptions.class).getProject(); - private static final String BIG_QUERY_DATASET_ID = "bigquery_managed_" + System.nanoTime(); + private static final String BIG_QUERY_DATASET_ID = + TEMP_DATASET_PREFIX + "managed_" + System.nanoTime(); private static final Clustering CLUSTERING = new Clustering().setFields(Arrays.asList("str")); diff --git a/sdks/python/apache_beam/io/gcp/bigquery_change_history_it_test.py b/sdks/python/apache_beam/io/gcp/bigquery_change_history_it_test.py index ef41fc393af7..8cc7c7bb9c34 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery_change_history_it_test.py +++ b/sdks/python/apache_beam/io/gcp/bigquery_change_history_it_test.py @@ -61,8 +61,8 @@ def setUpClass(cls): cls.args = cls.test_pipeline.get_full_options_as_args() cls.bq_wrapper = BigQueryWrapper() suffix = secrets.token_hex(4) - cls.dataset = f'beam_ch_src_{suffix}' - cls.temp_dataset = f'beam_ch_tmp_{suffix}' + cls.dataset = f'{bigquery_tools._TEMP_DATASET_PREFIX}ch_src_{suffix}' + cls.temp_dataset = f'{bigquery_tools._TEMP_DATASET_PREFIX}ch_tmp_{suffix}' cls.bq_wrapper.get_or_create_dataset(cls.project, cls.dataset) ds = cls.bq_wrapper.client.datasets.Get( bigquery.BigqueryDatasetsGetRequest( diff --git a/sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py b/sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py index fda4a3e9d520..7ae0ccc8e32f 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py +++ b/sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py @@ -1261,7 +1261,6 @@ def __call__(self): class BigQueryFileLoadsIT(unittest.TestCase): - BIG_QUERY_DATASET_ID = 'python_bq_file_loads_' BIG_QUERY_SCHEMA = ( '{"fields": [{"name": "name","type": "STRING"},' '{"name": "language","type": "STRING"}]}') @@ -1282,7 +1281,9 @@ def setUp(self): self.project = self.test_pipeline.get_option('project') self.dataset_id = '%s%d%s' % ( - self.BIG_QUERY_DATASET_ID, int(time.time()), secrets.token_hex(3)) + bigquery_tools._TEMP_DATASET_PREFIX, + int(time.time()), + secrets.token_hex(3)) self.bigquery_client = bigquery_tools.BigQueryWrapper() self.bigquery_client.get_or_create_dataset(self.project, self.dataset_id) self.output_table = "%s.output_table" % (self.dataset_id) diff --git a/sdks/python/apache_beam/io/gcp/bigquery_read_it_test.py b/sdks/python/apache_beam/io/gcp/bigquery_read_it_test.py index 013bcc5d8420..733b51a80521 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery_read_it_test.py +++ b/sdks/python/apache_beam/io/gcp/bigquery_read_it_test.py @@ -88,8 +88,6 @@ def datetime_to_utc(element): class BigQueryReadIntegrationTests(unittest.TestCase): - BIG_QUERY_DATASET_ID = 'python_read_table_' - @classmethod def setUpClass(cls): cls.test_pipeline = TestPipeline(is_integration_test=True) @@ -99,7 +97,9 @@ def setUpClass(cls): cls.bigquery_client = BigQueryWrapper() cls.dataset_id = '%s%d%s' % ( - cls.BIG_QUERY_DATASET_ID, int(time.time()), secrets.token_hex(3)) + bigquery_tools._TEMP_DATASET_PREFIX, + int(time.time()), + secrets.token_hex(3)) cls.bigquery_client.get_or_create_dataset(cls.project, cls.dataset_id) _LOGGER.info( "Created dataset %s in project %s", cls.dataset_id, cls.project) @@ -309,7 +309,6 @@ def test_table_schema_retrieve_with_direct_read(self): class ReadUsingStorageApiTests(BigQueryReadIntegrationTests): - BIG_QUERY_DATASET_ID = 'python_read_table_' TABLE_DATA = [{ 'number': 1, 'string': '你好', @@ -334,7 +333,9 @@ class ReadUsingStorageApiTests(BigQueryReadIntegrationTests): def setUpClass(cls): super(ReadUsingStorageApiTests, cls).setUpClass() cls.table_name = '%s%d%s' % ( - cls.BIG_QUERY_DATASET_ID, int(time.time()), secrets.token_hex(3)) + bigquery_tools._TEMP_DATASET_PREFIX, + int(time.time()), + secrets.token_hex(3)) cls._create_table(cls.table_name) table_id = '{}.{}'.format(cls.dataset_id, cls.table_name) diff --git a/sdks/python/apache_beam/io/gcp/bigquery_test.py b/sdks/python/apache_beam/io/gcp/bigquery_test.py index dcadee7f6a1a..874ae68362e8 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery_test.py +++ b/sdks/python/apache_beam/io/gcp/bigquery_test.py @@ -2953,15 +2953,15 @@ def test_file_loads(self): @unittest.skipIf(HttpError is None, 'GCP dependencies are not installed') class BigQueryFileLoadsIntegrationTests(unittest.TestCase): - BIG_QUERY_DATASET_ID = 'python_bq_file_loads_' - def setUp(self): self.test_pipeline = TestPipeline(is_integration_test=True) self.runner_name = type(self.test_pipeline.runner).__name__ self.project = self.test_pipeline.get_option('project') self.dataset_id = '%s%d%s' % ( - self.BIG_QUERY_DATASET_ID, int(time.time()), secrets.token_hex(3)) + bigquery_tools._TEMP_DATASET_PREFIX, + int(time.time()), + secrets.token_hex(3)) self.bigquery_client = bigquery_tools.BigQueryWrapper() self.bigquery_client.get_or_create_dataset(self.project, self.dataset_id) self.output_table = '%s.output_table' % (self.dataset_id) diff --git a/sdks/python/apache_beam/io/gcp/bigquery_tools.py b/sdks/python/apache_beam/io/gcp/bigquery_tools.py index 0d62ec5233c1..5cc3441e1719 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery_tools.py +++ b/sdks/python/apache_beam/io/gcp/bigquery_tools.py @@ -109,6 +109,9 @@ _DATASET_PATTERN = r'\w{1,1024}' _TABLE_PATTERN = r'[\p{L}\p{M}\p{N}\p{Pc}\p{Pd}\p{Zs}$]{1,1024}' +# For CI: temp dataset of this name are automatically deleted +_TEMP_DATASET_PREFIX = 'beam_temp_dataset_' + # TODO(https://github.com/apache/beam/issues/25946): Add support for # more Beam portable schema types as Python types BIGQUERY_TYPE_TO_PYTHON_TYPE = { diff --git a/sdks/python/apache_beam/ml/inference/vertex_ai_model_monitoring_v2_it_test.py b/sdks/python/apache_beam/ml/inference/vertex_ai_model_monitoring_v2_it_test.py index f09039497527..e2a43ec6a3fd 100644 --- a/sdks/python/apache_beam/ml/inference/vertex_ai_model_monitoring_v2_it_test.py +++ b/sdks/python/apache_beam/ml/inference/vertex_ai_model_monitoring_v2_it_test.py @@ -26,6 +26,7 @@ import pytest import apache_beam as beam +from apache_beam.io.gcp import bigquery_tools from apache_beam.ml.inference.base import ModelHandler from apache_beam.ml.inference.base import PredictionResult from apache_beam.ml.inference.base import RunInference @@ -70,7 +71,7 @@ class VertexAIModelMonitoringV2IntegrationTest(unittest.TestCase): def test_vertex_ai_model_monitoring_v2_batch_pipeline(self): test_pipeline = TestPipeline(is_integration_test=True) job_id = str(uuid.uuid4())[:8] - dataset_name = f"beam_mm_v2_{job_id}" + dataset_name = f"{bigquery_tools._TEMP_DATASET_PREFIX}mm_v2_{job_id}" predictions_table_name = "predictions" predictions_table_id = f"{_ENDPOINT_PROJECT}:{dataset_name}.{predictions_table_name}" display_name = f"beam-mm-v2-test-{job_id}" @@ -278,7 +279,7 @@ def test_vertex_ai_model_monitoring_v2_streaming_pipeline(self): test_pipeline = TestPipeline( is_integration_test=True, additional_pipeline_args=["--streaming"]) job_id = str(uuid.uuid4())[:8] - dataset_name = f"beam_mm_v2_str_{job_id}" + dataset_name = f"{bigquery_tools._TEMP_DATASET_PREFIX}mm_v2_str_{job_id}" predictions_table_name = "predictions" predictions_table_id = f"{_ENDPOINT_PROJECT}:{dataset_name}.{predictions_table_name}" display_name = f"beam-mm-v2-str-{job_id}"