Skip to content

Commit

Permalink
updated TypeHandler
Browse files Browse the repository at this point in the history
  • Loading branch information
@defector221
defector221 committed Nov 26, 2024
1 parent 270633c commit b5c105b
Showing 1 changed file with 22 additions and 21 deletions.
43 changes: 22 additions & 21 deletions ...rcedb/src/main/java/com/google/cloud/teleport/v2/templates/dml/CassandraDMLGenerator.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.PreparedStatement;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
Expand Down Expand Up @@ -63,7 +64,7 @@ public DMLGeneratorResponse getDMLStatement(DMLGeneratorRequest dmlGeneratorRequ
return new DMLGeneratorResponse("");
}

Map<String, String> pkColumnNameValues =
Map<String, Object> pkColumnNameValues =
getPkColumnValues(
spannerTable,
sourceTable,
Expand Down Expand Up @@ -100,8 +101,8 @@ private static DMLGeneratorResponse generateUpsertStatement(
SpannerTable spannerTable,
SourceTable sourceTable,
DMLGeneratorRequest dmlGeneratorRequest,
Map<String, String> pkColumnNameValues) {
Map<String, String> columnNameValues =
Map<String, Object> pkColumnNameValues) {
Map<String, Object> columnNameValues =
getColumnValues(
spannerTable,
sourceTable,
Expand All @@ -121,25 +122,25 @@ private static DMLGeneratorResponse generateUpsertStatement(
private static String getUpsertStatementCQL(
String tableName,
Set<String> primaryKeys,
Map<String, String> columnNameValues,
Map<String, String> pkcolumnNameValues) {
Map<String, Object> columnNameValues,
Map<String, Object> pkColumnNameValues) {

StringBuilder allColumns = new StringBuilder();
StringBuilder allValues = new StringBuilder();

// Process primary key columns
for (Map.Entry<String, String> entry : pkcolumnNameValues.entrySet()) {
for (Map.Entry<String, Object> entry : pkColumnNameValues.entrySet()) {
String colName = entry.getKey();
String colValue = entry.getValue();
Object colValue = entry.getValue();

allColumns.append(colName).append(", ");
allValues.append(colValue).append(", ");
}

// Process additional columns
for (Map.Entry<String, String> entry : columnNameValues.entrySet()) {
for (Map.Entry<String, Object> entry : columnNameValues.entrySet()) {
String colName = entry.getKey();
String colValue = entry.getValue();
Object colValue = entry.getValue();

allColumns.append(colName).append(", ");
allValues.append(colValue).append(", ");
Expand All @@ -158,18 +159,18 @@ private static String getUpsertStatementCQL(
}

private static String getDeleteStatementCQL(
String tableName, Map<String, String> pkcolumnNameValues) {
String tableName, Map<String, Object> pkColumnNameValues) {

StringBuilder deleteConditions = new StringBuilder();

// Process primary key columns for the WHERE clause
int index = 0;
for (Map.Entry<String, String> entry : pkcolumnNameValues.entrySet()) {
for (Map.Entry<String, Object> entry : pkColumnNameValues.entrySet()) {
String colName = entry.getKey();
String colValue = entry.getValue();
Object colValue = entry.getValue();

deleteConditions.append(colName).append(" = ").append(colValue);
if (index + 1 < pkcolumnNameValues.size()) {
if (index + 1 < pkColumnNameValues.size()) {
deleteConditions.append(" AND ");
}
index++;
Expand All @@ -179,14 +180,14 @@ private static String getDeleteStatementCQL(
return "DELETE FROM " + tableName + " WHERE " + deleteConditions + ";";
}

private static Map<String, String> getColumnValues(
private static Map<String, Object> getColumnValues(
SpannerTable spannerTable,
SourceTable sourceTable,
JSONObject newValuesJson,
JSONObject keyValuesJson,
String sourceDbTimezoneOffset
) {
Map<String, String> response = new HashMap<>();
Map<String, Object> response = new HashMap<>();

/*
Get all non-primary key col ids from source table
Expand Down Expand Up @@ -216,7 +217,7 @@ private static Map<String, String> getColumnValues(
continue;
}
String spannerColumnName = spannerColDef.getName();
String columnValue = "";
Object columnValue;
if (keyValuesJson.has(spannerColumnName)) {
// get the value based on Spanner and Source type
if (keyValuesJson.isNull(spannerColumnName)) {
Expand Down Expand Up @@ -245,14 +246,14 @@ private static Map<String, String> getColumnValues(
return response;
}

private static Map<String, String> getPkColumnValues(
private static Map<String, Object> getPkColumnValues(
SpannerTable spannerTable,
SourceTable sourceTable,
JSONObject newValuesJson,
JSONObject keyValuesJson,
String sourceDbTimezoneOffset
) {
Map<String, String> response = new HashMap<>();
Map<String, Object> response = new HashMap<>();
/*
Get all primary key col ids from source table
For each - get the corresponding column name from spanner Schema
Expand All @@ -277,7 +278,7 @@ private static Map<String, String> getPkColumnValues(
return null;
}
String spannerColumnName = spannerColDef.getName();
String columnValue = "";
Object columnValue;
if (keyValuesJson.has(spannerColumnName)) {
// get the value based on Spanner and Source type
if (keyValuesJson.isNull(spannerColumnName)) {
Expand Down Expand Up @@ -315,12 +316,12 @@ private static Map<String, String> getPkColumnValues(
return response;
}

private static String getMappedColumnValue(
private static Object getMappedColumnValue(
SpannerColumnDefinition spannerColDef,
SourceColumnDefinition sourceColDef,
JSONObject valuesJson,
String sourceDbTimezoneOffset) {
return (String) TypeHandler.getColumnValueByType(
return TypeHandler.getColumnValueByType(
spannerColDef,
sourceColDef,
valuesJson,
Expand Down

0 comments on commit b5c105b

Please sign in to comment.