/*
 * Copyright (C) 2021 Google LLC
 *
 * 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.google.cloud.teleport.v2.utils;

import com.google.cloud.teleport.v2.io.CdcJdbcIO.DataSourceConfiguration;
import com.google.cloud.teleport.v2.values.DatastreamRow;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** A set of Database Migration utilities to convert JSON data to DML. */
public class DatastreamToPostgresDML extends DatastreamToDML {

  private static final Logger LOG = LoggerFactory.getLogger(DatastreamToPostgresDML.class);

  private DatastreamToPostgresDML(DataSourceConfiguration config) {
    super(config);
  }

  public static DatastreamToPostgresDML of(DataSourceConfiguration config) {
    return new DatastreamToPostgresDML(config);
  }

  @Override
  public String getDefaultQuoteCharacter() {
    return "\"";
  }

  @Override
  public String getDeleteDmlStatement() {
    return "DELETE FROM {quoted_schema_name}.{quoted_table_name} WHERE {primary_key_kv_sql};";
  }

  @Override
  public String getUpsertDmlStatement() {
    return "INSERT INTO {quoted_schema_name}.{quoted_table_name} "
        + "({quoted_column_names}) VALUES ({column_value_sql}) "
        + "ON CONFLICT ({primary_key_names_sql}) DO UPDATE SET {column_kv_sql};";
  }

  @Override
  public String getInsertDmlStatement() {
    return "INSERT INTO {quoted_schema_name}.{quoted_table_name} "
        + "({quoted_column_names}) VALUES ({column_value_sql});";
  }

  @Override
  public String getTargetCatalogName(DatastreamRow row) {
    return "";
  }

  @Override
  public String getTargetSchemaName(DatastreamRow row) {
    String schemaName = row.getSchemaName();
    return cleanSchemaName(schemaName);
  }

  @Override
  public String getTargetTableName(DatastreamRow row) {
    String tableName = row.getTableName();
    return cleanTableName(tableName);
  }

  @Override
  public String cleanDataTypeValueSql(
      String columnValue, String columnName, Map<String, String> tableSchema) {
    String dataType = tableSchema.get(columnName);
    if (dataType == null) {
      return columnValue;
    }
    switch (dataType.toUpperCase()) {
      case "DECIMAL":
        if (columnValue.equals("")) {
          return getNullValueSql();
        }
        break;
    }
    return columnValue;
  }
}

@@ -2696,50 +2696,386 @@
 se %22
-DECIMAL%22:%0A        if (columnValue.equals(%22
+INT2%22:%0A      case %22INT4%22:%0A      case %22INT8%22:%0A      case %22FLOAT4%22:%0A      case %22FLOAT8%22:%0A      case %22SMALLINT%22:%0A      case %22INTEGER%22:%0A      case %22BIGINT%22:%0A      case %22DECIMAL%22:%0A      case %22NUMERIC%22:%0A      case %22REAL%22:%0A      case %22DOUBLE PRECISION%22:%0A      case %22SMALLSERIAL%22:%0A      case %22SERIAL%22:%0A      case %22BIGSERIAL%22:%0A        if (columnValue.equals(%22%22) %7C%7C columnValue.equals(%22''
 %22)) 

clean empty or empty string numeric data to null PiperOrigin-RevId: 448769626