openecomp-be/tools/zusammen-tools/src/main/java/org/openecomp/core/tools/importinfo/ImportSingleTable.java

   1 package org.openecomp.core.tools.importinfo;
   2
   3 import static org.openecomp.core.tools.exportinfo.ExportDataCommand.NULL_REPRESENTATION;
   4
   5 import com.datastax.driver.core.BoundStatement;
   6 import com.datastax.driver.core.DataType.Name;
   7 import com.datastax.driver.core.PreparedStatement;
   8 import com.datastax.driver.core.Session;
   9 import com.google.common.collect.ImmutableMap;
  10 import com.google.common.collect.ImmutableMap.Builder;
  11 import com.google.common.collect.Sets;
  12 import org.apache.commons.lang3.StringUtils;
  13 import org.codehaus.jackson.map.ObjectMapper;
  14 import org.openecomp.core.nosqldb.impl.cassandra.CassandraSessionFactory;
  15 import org.openecomp.core.tools.exportinfo.ExportDataCommand;
  16 import org.openecomp.core.tools.model.ColumnDefinition;
  17 import org.openecomp.core.tools.model.TableData;
  18 import org.openecomp.core.tools.util.Utils;
  19 import org.openecomp.sdc.logging.api.Logger;
  20 import org.openecomp.sdc.logging.api.LoggerFactory;
  21
  22 import java.io.IOException;
  23 import java.nio.ByteBuffer;
  24 import java.nio.file.Path;
  25 import java.util.*;
  26 import java.util.stream.Collectors;
  27
  28 public class ImportSingleTable {
  29
  30     private static final Logger logger = LoggerFactory.getLogger(ImportSingleTable.class);
  31
  32     private static final String INSERT_INTO = "INSERT INTO ";
  33     private static final String VALUES = " VALUES ";
  34     private static final Map<String, PreparedStatement> statementsCache = new HashMap<>();
  35
  36     public void importFile(Path file) {
  37         try {
  38             ObjectMapper objectMapper = new ObjectMapper();
  39             TableData tableData = objectMapper.readValue(file.toFile(), TableData.class);
  40             Session session = CassandraSessionFactory.getSession();
  41             PreparedStatement ps = getPrepareStatement(tableData, session);
  42             tableData.rows.forEach(row -> executeQuery(session, ps, tableData.definitions, row));
  43         } catch (IOException e) {
  44             Utils.logError(logger, e);
  45         }
  46
  47     }
  48
  49     private PreparedStatement getPrepareStatement(TableData tableData, Session session) {
  50         String query = createQuery(tableData);
  51         if (statementsCache.containsKey(query)) {
  52             return statementsCache.get(query);
  53         }
  54         PreparedStatement preparedStatement = session.prepare(query);
  55         statementsCache.put(query, preparedStatement);
  56         return preparedStatement;
  57     }
  58
  59     private void executeQuery(Session session, PreparedStatement ps, List<ColumnDefinition> definitions, List<String> rows) {
  60         BoundStatement bind = ps.bind();
  61         for (int i = 0; i < definitions.size(); i++) {
  62             ColumnDefinition columnDefinition = definitions.get(i);
  63             String rowData = rows.get(i);
  64             Name name = dataTypesMap.get(columnDefinition.getType());
  65             handleByType(bind, i, rowData, name);
  66         }
  67         session.execute(bind);
  68     }
  69
  70     private void handleByType(BoundStatement bind, int i, String rowData, Name name) {
  71         switch (name) {
  72             case VARCHAR:
  73             case TEXT:
  74             case ASCII:
  75                 String string = new String(Base64.getDecoder().decode(rowData));
  76                 bind.setString(i, NULL_REPRESENTATION.equals(string) ? null : string);
  77                 break;
  78             case BLOB:
  79                 bind.setBytes(i, ByteBuffer.wrap(Base64.getDecoder().decode(rowData.getBytes())));
  80                 break;
  81             case TIMESTAMP:
  82                 if (StringUtils.isEmpty(rowData)){
  83                     bind.setTimestamp(i, null);
  84                 } else {
  85                     bind.setTimestamp(i, new Date(Long.parseLong(rowData)));
  86                 }
  87                 break;
  88             case BOOLEAN:
  89                 bind.setBool(i, Boolean.parseBoolean(rowData));
  90                 break;
  91             case COUNTER:
  92                 bind.setLong(i, Long.parseLong(rowData));
  93                 break;
  94             case INT:
  95                 bind.setInt(i, Integer.parseInt(rowData));
  96                 break;
  97             case FLOAT:
  98                 bind.setFloat(i, Float.parseFloat(rowData));
  99                 break;
 100             case SET:
 101                 byte[] decoded = Base64.getDecoder().decode(rowData);
 102                 String decodedStr = new String(decoded);
 103                 if (!StringUtils.isEmpty(decodedStr)) {
 104                     String[] splitted = decodedStr.split(ExportDataCommand.JOIN_DELIMITER_SPLITTER);
 105                     Set set = Sets.newHashSet(splitted);
 106                     set.remove("");
 107                     bind.setSet(i, set);
 108                 } else {
 109                     bind.setSet(i, null);
 110                 }
 111                 break;
 112             case MAP:
 113                 byte[] decodedMap = Base64.getDecoder().decode(rowData);
 114                 String mapStr = new String(decodedMap);
 115                 if (!StringUtils.isEmpty(mapStr)) {
 116                     String[] splittedMap = mapStr.split(ExportDataCommand.JOIN_DELIMITER_SPLITTER);
 117                     Map<String, String> map = new HashMap<>();
 118                     for (String keyValue : splittedMap) {
 119                         String[] split = keyValue.split(ExportDataCommand.MAP_DELIMITER_SPLITTER);
 120                         map.put(split[0], split[1]);
 121                     }
 122                     bind.setMap(i, map);
 123                 } else {
 124                     bind.setMap(i, null);
 125                 }
 126                 break;
 127             default:
 128                 throw new UnsupportedOperationException("Name is not supported :" + name);
 129
 130         }
 131     }
 132
 133     private String createQuery(TableData tableData) {
 134         ColumnDefinition def = tableData.definitions.iterator().next();
 135         StringBuilder sb = new StringBuilder();
 136         sb.append(INSERT_INTO).append(def.getKeyspace()).append(".").append(def.getTable());
 137         sb.append(tableData.definitions.stream().map(ColumnDefinition::getName).collect(Collectors.joining(" , ", " ( ", " ) ")));
 138         sb.append(VALUES).append(tableData.definitions.stream().map(definition -> "?").collect(Collectors.joining(" , ", " ( ", " ) "))).append(";");
 139         return sb.toString();
 140     }
 141
 142     public static final ImmutableMap<String, Name> dataTypesMap;
 143
 144     static {
 145         Builder<String, Name> builder = ImmutableMap.builder();
 146         Name[] values = Name.values();
 147         for (Name name : values) {
 148             builder.put(name.name().toLowerCase(), name);
 149         }
 150         dataTypesMap = builder.build();
 151     }
 152
 153 }