@Test public void insertTest() { try { String insertKey = "user0"; HashMap<String, ByteIterator> insertMap = insertRow(insertKey); ResultSet resultSet = jdbcConnection.prepareStatement( String.format("SELECT * FROM %s", TABLE_NAME) ).executeQuery(); assertTrue(resultSet.next()); assertEquals(resultSet.getString(KEY_FIELD), insertKey); for (int i = 0; i < 3; i++) { assertEquals(resultSet.getString(FIELD_PREFIX + i), insertMap.get(FIELD_PREFIX + i).toString()); } assertFalse(resultSet.next()); resultSet.close(); } catch (SQLException e) { e.printStackTrace(); fail("Failed insertTest"); } }
@Override public Status insert(String tableName, String key, Map<String, ByteIterator> values) { try { int numFields = values.size(); OrderedFieldInfo fieldInfo = getFieldInfo(values); StatementType type = new StatementType(StatementType.Type.INSERT, tableName, numFields, fieldInfo.getFieldKeys(), getShardIndexByKey(key)); PreparedStatement insertStatement = cachedStatements.get(type); if (insertStatement == null) { insertStatement = createAndCacheInsertStatement(type, key); } insertStatement.setString(1, key); int index = 2; for (String value: fieldInfo.getFieldValues()) { insertStatement.setString(index++, value); } if (batchUpdates) { insertStatement.addBatch(); if (batchSize > 0) { if (++numRowsInBatch % batchSize == 0) { int[] results = insertStatement.executeBatch(); for (int r : results) { if (r != 1 && r != -2) { return Status.ERROR; } } if (!autoCommit) { getShardConnectionByKey(key).commit(); } return Status.OK; } } return Status.BATCHED_OK; } else { int result = insertStatement.executeUpdate(); if (!autoCommit) { if (batchSize > 0) { if (++numRowsInBatch % batchSize == 0) { getShardConnectionByKey(key).commit(); } return Status.OK; } else { getShardConnectionByKey(key).commit(); } } if (result == 1) { return Status.OK; } } return Status.UNEXPECTED_STATE; } catch (SQLException e) { System.err.println("Error in processing insert to table: " + tableName + e); return Status.ERROR; } }
JdbcDBClient extends DB { @Override public Status insert(String tableName, String key, Map<String, ByteIterator> values) { try { int numFields = values.size(); OrderedFieldInfo fieldInfo = getFieldInfo(values); StatementType type = new StatementType(StatementType.Type.INSERT, tableName, numFields, fieldInfo.getFieldKeys(), getShardIndexByKey(key)); PreparedStatement insertStatement = cachedStatements.get(type); if (insertStatement == null) { insertStatement = createAndCacheInsertStatement(type, key); } insertStatement.setString(1, key); int index = 2; for (String value: fieldInfo.getFieldValues()) { insertStatement.setString(index++, value); } if (batchUpdates) { insertStatement.addBatch(); if (batchSize > 0) { if (++numRowsInBatch % batchSize == 0) { int[] results = insertStatement.executeBatch(); for (int r : results) { if (r != 1 && r != -2) { return Status.ERROR; } } if (!autoCommit) { getShardConnectionByKey(key).commit(); } return Status.OK; } } return Status.BATCHED_OK; } else { int result = insertStatement.executeUpdate(); if (!autoCommit) { if (batchSize > 0) { if (++numRowsInBatch % batchSize == 0) { getShardConnectionByKey(key).commit(); } return Status.OK; } else { getShardConnectionByKey(key).commit(); } } if (result == 1) { return Status.OK; } } return Status.UNEXPECTED_STATE; } catch (SQLException e) { System.err.println("Error in processing insert to table: " + tableName + e); return Status.ERROR; } } }
JdbcDBClient extends DB { @Override public Status insert(String tableName, String key, Map<String, ByteIterator> values) { try { int numFields = values.size(); OrderedFieldInfo fieldInfo = getFieldInfo(values); StatementType type = new StatementType(StatementType.Type.INSERT, tableName, numFields, fieldInfo.getFieldKeys(), getShardIndexByKey(key)); PreparedStatement insertStatement = cachedStatements.get(type); if (insertStatement == null) { insertStatement = createAndCacheInsertStatement(type, key); } insertStatement.setString(1, key); int index = 2; for (String value: fieldInfo.getFieldValues()) { insertStatement.setString(index++, value); } if (batchUpdates) { insertStatement.addBatch(); if (batchSize > 0) { if (++numRowsInBatch % batchSize == 0) { int[] results = insertStatement.executeBatch(); for (int r : results) { if (r != 1 && r != -2) { return Status.ERROR; } } if (!autoCommit) { getShardConnectionByKey(key).commit(); } return Status.OK; } } return Status.BATCHED_OK; } else { int result = insertStatement.executeUpdate(); if (!autoCommit) { if (batchSize > 0) { if (++numRowsInBatch % batchSize == 0) { getShardConnectionByKey(key).commit(); } return Status.OK; } else { getShardConnectionByKey(key).commit(); } } if (result == 1) { return Status.OK; } } return Status.UNEXPECTED_STATE; } catch (SQLException e) { System.err.println("Error in processing insert to table: " + tableName + e); return Status.ERROR; } }  }
JdbcDBClient extends DB { @Override public Status insert(String tableName, String key, Map<String, ByteIterator> values) { try { int numFields = values.size(); OrderedFieldInfo fieldInfo = getFieldInfo(values); StatementType type = new StatementType(StatementType.Type.INSERT, tableName, numFields, fieldInfo.getFieldKeys(), getShardIndexByKey(key)); PreparedStatement insertStatement = cachedStatements.get(type); if (insertStatement == null) { insertStatement = createAndCacheInsertStatement(type, key); } insertStatement.setString(1, key); int index = 2; for (String value: fieldInfo.getFieldValues()) { insertStatement.setString(index++, value); } if (batchUpdates) { insertStatement.addBatch(); if (batchSize > 0) { if (++numRowsInBatch % batchSize == 0) { int[] results = insertStatement.executeBatch(); for (int r : results) { if (r != 1 && r != -2) { return Status.ERROR; } } if (!autoCommit) { getShardConnectionByKey(key).commit(); } return Status.OK; } } return Status.BATCHED_OK; } else { int result = insertStatement.executeUpdate(); if (!autoCommit) { if (batchSize > 0) { if (++numRowsInBatch % batchSize == 0) { getShardConnectionByKey(key).commit(); } return Status.OK; } else { getShardConnectionByKey(key).commit(); } } if (result == 1) { return Status.OK; } } return Status.UNEXPECTED_STATE; } catch (SQLException e) { System.err.println("Error in processing insert to table: " + tableName + e); return Status.ERROR; } }  @Override void init(); @Override void cleanup(); @Override Status read(String tableName, String key, Set<String> fields, Map<String, ByteIterator> result); @Override Status readLog(String table, int logcount); @Override Status readMeta(String tableName, String cond, String keymatch, Vector<HashMap<String, ByteIterator>> result); @Override Status scan(String tableName, String startKey, int recordcount, Set<String> fields,
                     Vector<HashMap<String, ByteIterator>> result); @Override Status update(String tableName, String key, Map<String, ByteIterator> values); @Override Status updateMeta(String table, String condition, String keymatch, String fieldname, String metadatavalue); @Override Status insert(String tableName, String key, Map<String, ByteIterator> values); @Override Status delete(String tableName, String key); @Override Status deleteMeta(String table, String condition, String keymatch); @Override Status verifyTTL(String table, long recordcount); @Override Status insertTTL(String table, String key,
                         Map<String, ByteIterator> values, int ttl); }
JdbcDBClient extends DB { @Override public Status insert(String tableName, String key, Map<String, ByteIterator> values) { try { int numFields = values.size(); OrderedFieldInfo fieldInfo = getFieldInfo(values); StatementType type = new StatementType(StatementType.Type.INSERT, tableName, numFields, fieldInfo.getFieldKeys(), getShardIndexByKey(key)); PreparedStatement insertStatement = cachedStatements.get(type); if (insertStatement == null) { insertStatement = createAndCacheInsertStatement(type, key); } insertStatement.setString(1, key); int index = 2; for (String value: fieldInfo.getFieldValues()) { insertStatement.setString(index++, value); } if (batchUpdates) { insertStatement.addBatch(); if (batchSize > 0) { if (++numRowsInBatch % batchSize == 0) { int[] results = insertStatement.executeBatch(); for (int r : results) { if (r != 1 && r != -2) { return Status.ERROR; } } if (!autoCommit) { getShardConnectionByKey(key).commit(); } return Status.OK; } } return Status.BATCHED_OK; } else { int result = insertStatement.executeUpdate(); if (!autoCommit) { if (batchSize > 0) { if (++numRowsInBatch % batchSize == 0) { getShardConnectionByKey(key).commit(); } return Status.OK; } else { getShardConnectionByKey(key).commit(); } } if (result == 1) { return Status.OK; } } return Status.UNEXPECTED_STATE; } catch (SQLException e) { System.err.println("Error in processing insert to table: " + tableName + e); return Status.ERROR; } }  @Override void init(); @Override void cleanup(); @Override Status read(String tableName, String key, Set<String> fields, Map<String, ByteIterator> result); @Override Status readLog(String table, int logcount); @Override Status readMeta(String tableName, String cond, String keymatch, Vector<HashMap<String, ByteIterator>> result); @Override Status scan(String tableName, String startKey, int recordcount, Set<String> fields,
                     Vector<HashMap<String, ByteIterator>> result); @Override Status update(String tableName, String key, Map<String, ByteIterator> values); @Override Status updateMeta(String table, String condition, String keymatch, String fieldname, String metadatavalue); @Override Status insert(String tableName, String key, Map<String, ByteIterator> values); @Override Status delete(String tableName, String key); @Override Status deleteMeta(String table, String condition, String keymatch); @Override Status verifyTTL(String table, long recordcount); @Override Status insertTTL(String table, String key,
                         Map<String, ByteIterator> values, int ttl); static final String DRIVER_CLASS; static final String CONNECTION_URL; static final String CONNECTION_USER; static final String CONNECTION_PASSWD; static final String DB_BATCH_SIZE; static final String JDBC_FETCH_SIZE; static final String JDBC_AUTO_COMMIT; static final String JDBC_BATCH_UPDATES; static final String FIELD_COUNT_PROPERTY; static final String FIELD_COUNT_PROPERTY_DEFAULT; static final String NULL_VALUE; static final String PRIMARY_KEY; static final String COLUMN_PREFIX; }