<commit_before>/*
 * Copyright © 2019 <NAME>, 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 io.cdap.delta.bigquery;

import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.DatasetId;
import com.google.cloud.bigquery.DatasetInfo;
import com.google.cloud.bigquery.FieldValue;
import com.google.cloud.bigquery.FieldValueList;
import com.google.cloud.bigquery.FormatOptions;
import com.google.cloud.bigquery.Job;
import com.google.cloud.bigquery.JobId;
import com.google.cloud.bigquery.JobInfo;
import com.google.cloud.bigquery.LoadJobConfiguration;
import com.google.cloud.bigquery.QueryJobConfiguration;
import com.google.cloud.bigquery.StandardTableDefinition;
import com.google.cloud.bigquery.Table;
import com.google.cloud.bigquery.TableDefinition;
import com.google.cloud.bigquery.TableId;
import com.google.cloud.bigquery.TableInfo;
import com.google.cloud.bigquery.TableResult;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.Bucket;
import com.google.cloud.storage.Storage;
import com.google.common.annotations.VisibleForTesting;
import io.cdap.cdap.api.data.schema.Schema;
import io.cdap.delta.api.DDLEvent;
import io.cdap.delta.api.DMLEvent;
import io.cdap.delta.api.DeltaFailureException;
import io.cdap.delta.api.DeltaTargetContext;
import io.cdap.delta.api.EventConsumer;
import io.cdap.delta.api.Offset;
import io.cdap.delta.api.ReplicationError;
import io.cdap.delta.api.Sequenced;
import net.jodah.failsafe.Failsafe;
import net.jodah.failsafe.FailsafeException;
import net.jodah.failsafe.RetryPolicy;
import net.jodah.failsafe.TimeoutExceededException;
import net.jodah.failsafe.function.ContextualRunnable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Consumes change events and applies them to BigQuery.
 *
 * Writes to BigQuery in three steps.
 *
 * Step 1 - Write a batch of changes to GCS
 *
 * Each batch of changes is written to GCS as an object with path:
 *
 *   [staging bucket]/cdap/cdc/[app name]/[table id]/[batch id]
 *
 * Batch id is the timestamp that the first event in the batch was processed.
 * The size of the batch is determined through configuration.
 * There is a maximum number of rows to include in each batch and a maximum amount of time to wait in between batches.
 * Each object is written in avro format and contains the columns in the destination table plus two additional columns:
 *   _op: CREATE | UPDATE | DELETE
 *   _batch_id: the batch id
 *
 * Changes in the batch do not span across a DDL event, so they are guaranteed to conform to the same schema.
 * Failure scenarios are:
 *
 *   1. The program dies after the object is written, but before the offset is persisted.
 *      When the program starts up again, events for the batch will be replayed.
 *      The consumer will not know which events are duplicates, so duplicate events will be written out to GCS.
 *      This will not matter because of the behavior of later steps.
 *   2. The program dies before the object is written, which is always before the offset is persisted.
 *      In this case, nothing was ever persisted to GCS and everything behaves as if it was the first time
 *      the events were seen.
 *   3. The write to GCS fails for some reason. For example, permissions were revoked, quota was hit,
 *      there was a temporary outage, etc. In this scenario, the write will be repeatedly retried until it
 *      succeeds. It may need manual intervention to succeed.
 *
 * Step 2 - Load data from GCS into staging BigQuery table
 *
 * This step happens after the offset from Step 1 is successfully persisted. This will load the object
 * into a staging table in BigQuery. The staging table has the same schema as the rows in the GCS object.
 * It is clustered on _batch_id in order to make reads and deletes on the _batch_id efficient.
 * The job id for the load is of the form [app name]_stage_[dataset]_[table]_[batch id]_[retry num].
 * Failure scenarios are:
 *
 *   1. The load job fails for some reason. For example, permissions were revoked, quota was hit, temporary outage, etc.
 *      The load will be repeatedly retried until is succeeds. It may need manual intervention to succeed.
 *   2. The program dies. When the program starts up again, events will be replayed from the last committed offset.
 *
 * Step 3 - Merge a batch of data from the staging BigQuery table into the target table
 *
 * This step happens after the load job to the staging table has succeeded. The consumer runs a merge query of the form:
 *
 *   MERGE [dataset].[target table] as T
 *   USING (SELECT * FROM [dataset].[staging table] WHERE _batch_id = [batch id]) as S
 *     ON [row equality condition]
 *   WHEN MATCHED AND S._OP = "DELETE"
 *     DELETE
 *   WHEN MATCHED AND S._OP = "UPDATE"
 *     UPDATE(...)
 *   WHEN NOT MATCHED AND S._OP = "INSERT"
 *     INSERT(...)
 *     VALUES(...)
 *
 * The job id is of the form [app name]_merge_[dataset]_[table]_[batch id]_[retry_num].
 * This query ensures that it does not matter if there are duplicate events in the batch objects on GCS.
 * Duplicate inserts and deletes will not match and be ignored.
 * Duplicate updates will update the target row to be the same that it already is.
 * Once the job succeeds, the corresponding GCS object is deleted and the offset of the latest event is committed.
 * Failure scenarios are:
 *
 *   1. The merge query fails for some reason. The consumer will retry until it succeeds.
 *      It may need manual intervention to succeed.
 *   2. The program dies. Events are replayed from the last committed offset when the program starts back up.
 *   3. The GCS delete fails. The error is logged, but the consumer proceeds on.
 *      Manual deletion of the object is required.
 *
 */
public class BigQueryEventConsumer implements EventConsumer {
  private static final Logger LOG = LoggerFactory.getLogger(BigQueryEventConsumer.class);
  static final int MAX_LENGTH = 1024;
  // according to big query dataset and table naming convention, valid name should only contain letters (upper or
  // lower case), numbers, and underscores
  static final String VALID_NAME_REGEX = "[\\w]+";
  private static final String INVALID_NAME_REGEX = "[^\\w]+";

  private final DeltaTargetContext context;
  private final BigQuery bigQuery;
  private final int batchMaxRows;
  private final int batchMaxSecondsElapsed;
  private final String stagingTablePrefix;
  private final MultiGCSWriter gcsWriter;
  private final Bucket bucket;
  private final String project;
  private final RetryPolicy<Object> commitRetryPolicy;<commit_msg>change final variable to private.<commit_after>/*
 * Copyright © 2019 Cask Data, 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 io.cdap.delta.bigquery;

import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.DatasetId;
import com.google.cloud.bigquery.DatasetInfo;
import com.google.cloud.bigquery.FieldValue;
import com.google.cloud.bigquery.FieldValueList;
import com.google.cloud.bigquery.FormatOptions;
import com.google.cloud.bigquery.Job;
import com.google.cloud.bigquery.JobId;
import com.google.cloud.bigquery.JobInfo;
import com.google.cloud.bigquery.LoadJobConfiguration;
import com.google.cloud.bigquery.QueryJobConfiguration;
import com.google.cloud.bigquery.StandardTableDefinition;
import com.google.cloud.bigquery.Table;
import com.google.cloud.bigquery.TableDefinition;
import com.google.cloud.bigquery.TableId;
import com.google.cloud.bigquery.TableInfo;
import com.google.cloud.bigquery.TableResult;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.Bucket;
import com.google.cloud.storage.Storage;
import com.google.common.annotations.VisibleForTesting;
import io.cdap.cdap.api.data.schema.Schema;
import io.cdap.delta.api.DDLEvent;
import io.cdap.delta.api.DMLEvent;
import io.cdap.delta.api.DeltaFailureException;
import io.cdap.delta.api.DeltaTargetContext;
import io.cdap.delta.api.EventConsumer;
import io.cdap.delta.api.Offset;
import io.cdap.delta.api.ReplicationError;
import io.cdap.delta.api.Sequenced;
import net.jodah.failsafe.Failsafe;
import net.jodah.failsafe.FailsafeException;
import net.jodah.failsafe.RetryPolicy;
import net.jodah.failsafe.TimeoutExceededException;
import net.jodah.failsafe.function.ContextualRunnable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Consumes change events and applies them to BigQuery.
 *
 * Writes to BigQuery in three steps.
 *
 * Step 1 - Write a batch of changes to GCS
 *
 * Each batch of changes is written to GCS as an object with path:
 *
 *   [staging bucket]/cdap/cdc/[app name]/[table id]/[batch id]
 *
 * Batch id is the timestamp that the first event in the batch was processed.
 * The size of the batch is determined through configuration.
 * There is a maximum number of rows to include in each batch and a maximum amount of time to wait in between batches.
 * Each object is written in avro format and contains the columns in the destination table plus two additional columns:
 *   _op: CREATE | UPDATE | DELETE
 *   _batch_id: the batch id
 *
 * Changes in the batch do not span across a DDL event, so they are guaranteed to conform to the same schema.
 * Failure scenarios are:
 *
 *   1. The program dies after the object is written, but before the offset is persisted.
 *      When the program starts up again, events for the batch will be replayed.
 *      The consumer will not know which events are duplicates, so duplicate events will be written out to GCS.
 *      This will not matter because of the behavior of later steps.
 *   2. The program dies before the object is written, which is always before the offset is persisted.
 *      In this case, nothing was ever persisted to GCS and everything behaves as if it was the first time
 *      the events were seen.
 *   3. The write to GCS fails for some reason. For example, permissions were revoked, quota was hit,
 *      there was a temporary outage, etc. In this scenario, the write will be repeatedly retried until it
 *      succeeds. It may need manual intervention to succeed.
 *
 * Step 2 - Load data from GCS into staging BigQuery table
 *
 * This step happens after the offset from Step 1 is successfully persisted. This will load the object
 * into a staging table in BigQuery. The staging table has the same schema as the rows in the GCS object.
 * It is clustered on _batch_id in order to make reads and deletes on the _batch_id efficient.
 * The job id for the load is of the form [app name]_stage_[dataset]_[table]_[batch id]_[retry num].
 * Failure scenarios are:
 *
 *   1. The load job fails for some reason. For example, permissions were revoked, quota was hit, temporary outage, etc.
 *      The load will be repeatedly retried until is succeeds. It may need manual intervention to succeed.
 *   2. The program dies. When the program starts up again, events will be replayed from the last committed offset.
 *
 * Step 3 - Merge a batch of data from the staging BigQuery table into the target table
 *
 * This step happens after the load job to the staging table has succeeded. The consumer runs a merge query of the form:
 *
 *   MERGE [dataset].[target table] as T
 *   USING (SELECT * FROM [dataset].[staging table] WHERE _batch_id = [batch id]) as S
 *     ON [row equality condition]
 *   WHEN MATCHED AND S._OP = "DELETE"
 *     DELETE
 *   WHEN MATCHED AND S._OP = "UPDATE"
 *     UPDATE(...)
 *   WHEN NOT MATCHED AND S._OP = "INSERT"
 *     INSERT(...)
 *     VALUES(...)
 *
 * The job id is of the form [app name]_merge_[dataset]_[table]_[batch id]_[retry_num].
 * This query ensures that it does not matter if there are duplicate events in the batch objects on GCS.
 * Duplicate inserts and deletes will not match and be ignored.
 * Duplicate updates will update the target row to be the same that it already is.
 * Once the job succeeds, the corresponding GCS object is deleted and the offset of the latest event is committed.
 * Failure scenarios are:
 *
 *   1. The merge query fails for some reason. The consumer will retry until it succeeds.
 *      It may need manual intervention to succeed.
 *   2. The program dies. Events are replayed from the last committed offset when the program starts back up.
 *   3. The GCS delete fails. The error is logged, but the consumer proceeds on.
 *      Manual deletion of the object is required.
 *
 */
public class BigQueryEventConsumer implements EventConsumer {
  private static final Logger LOG = LoggerFactory.getLogger(BigQueryEventConsumer.class);
  private static final int MAX_LENGTH = 1024;
  // according to big query dataset and table naming convention, valid name should only contain letters (upper or
  // lower case), numbers, and underscores
  private static final String VALID_NAME_REGEX = "[\\w]+";
  private static final String INVALID_NAME_REGEX = "[^\\w]+";

  private final DeltaTargetContext context;
  private final BigQuery bigQuery;
  private final int batchMaxRows;
  private final int batchMaxSecondsElapsed;
  private final String stagingTablePrefix;
  private final MultiGCSWriter gcsWriter;
  private final Bucket bucket;
  private final String project;
  private final RetryPolicy<Object> commitRetryPolicy;
{'fraction_non_alphanumeric': 0.0742752437667692, 'fraction_numerical': 0.003402918657155945, 'mean_word_length': 4.045229448662925, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '936865', 'n_tokens_mistral': 4320, 'n_tokens_neox': 4186, 'n_words': 1812}