You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by bl...@apache.org on 2014/02/19 16:02:59 UTC
[1/3] TAJO-530: Fix warnings in tajo-catalog. (jaehwa)
Repository: incubator-tajo
Updated Branches:
refs/heads/master c58622775 -> 413a4b3fc
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java
index 984ba6a..e05e0ae 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java
@@ -25,6 +25,7 @@ import com.google.common.collect.Maps;
import org.apache.hadoop.conf.Configuration;
import org.apache.tajo.catalog.CatalogUtil;
import org.apache.tajo.catalog.FunctionDesc;
+import org.apache.tajo.catalog.exception.CatalogException;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.proto.CatalogProtos.IndexDescProto;
@@ -56,7 +57,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#addTable(TableDesc)
*/
@Override
- public void addTable(CatalogProtos.TableDescProto desc) throws IOException {
+ public void addTable(CatalogProtos.TableDescProto desc) throws CatalogException {
synchronized(tables) {
String tableId = desc.getId().toLowerCase();
tables.put(tableId, desc);
@@ -67,7 +68,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#existTable(java.lang.String)
*/
@Override
- public boolean existTable(String name) throws IOException {
+ public boolean existTable(String name) throws CatalogException {
synchronized(tables) {
String tableId = name.toLowerCase();
return tables.containsKey(tableId);
@@ -78,7 +79,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#deleteTable(java.lang.String)
*/
@Override
- public void deleteTable(String name) throws IOException {
+ public void deleteTable(String name) throws CatalogException {
synchronized(tables) {
String tableId = name.toLowerCase();
tables.remove(tableId);
@@ -89,7 +90,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#getTable(java.lang.String)
*/
@Override
- public CatalogProtos.TableDescProto getTable(String name) throws IOException {
+ public CatalogProtos.TableDescProto getTable(String name) throws CatalogException {
String tableId = name.toLowerCase();
CatalogProtos.TableDescProto unqualified = tables.get(tableId);
if(unqualified == null)
@@ -105,69 +106,69 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#getAllTableNames()
*/
@Override
- public List<String> getAllTableNames() throws IOException {
+ public List<String> getAllTableNames() throws CatalogException {
return new ArrayList<String>(tables.keySet());
}
@Override
- public void addPartitionMethod(CatalogProtos.PartitionMethodProto partitionMethodProto) throws IOException {
- throw new IOException("not supported!");
+ public void addPartitionMethod(CatalogProtos.PartitionMethodProto partitionMethodProto) throws CatalogException {
+ throw new RuntimeException("not supported!");
}
@Override
- public CatalogProtos.PartitionMethodProto getPartitionMethod(String tableName) throws IOException {
+ public CatalogProtos.PartitionMethodProto getPartitionMethod(String tableName) throws CatalogException {
String tableId = tableName.toLowerCase();
CatalogProtos.TableDescProto table = tables.get(tableId);
return (table != null && table.hasPartition()) ? table.getPartition() : null;
}
@Override
- public boolean existPartitionMethod(String tableName) throws IOException {
+ public boolean existPartitionMethod(String tableName) throws CatalogException {
String tableId = tableName.toLowerCase();
CatalogProtos.TableDescProto table = tables.get(tableId);
return (table != null && table.hasPartition());
}
@Override
- public void delPartitionMethod(String tableName) throws IOException {
- throw new IOException("not supported!");
+ public void delPartitionMethod(String tableName) throws CatalogException {
+ throw new RuntimeException("not supported!");
}
@Override
- public void addPartitions(CatalogProtos.PartitionsProto partitionDescList) throws IOException {
- throw new IOException("not supported!");
+ public void addPartitions(CatalogProtos.PartitionsProto partitionDescList) throws CatalogException {
+ throw new RuntimeException("not supported!");
}
@Override
- public void addPartition(CatalogProtos.PartitionDescProto partitionDesc) throws IOException {
- throw new IOException("not supported!");
+ public void addPartition(CatalogProtos.PartitionDescProto partitionDesc) throws CatalogException {
+ throw new RuntimeException("not supported!");
}
@Override
- public CatalogProtos.PartitionsProto getPartitions(String tableName) throws IOException {
- throw new IOException("not supported!");
+ public CatalogProtos.PartitionsProto getPartitions(String tableName) throws CatalogException {
+ throw new RuntimeException("not supported!");
}
@Override
- public CatalogProtos.PartitionDescProto getPartition(String partitionName) throws IOException {
- throw new IOException("not supported!");
+ public CatalogProtos.PartitionDescProto getPartition(String partitionName) throws CatalogException {
+ throw new RuntimeException("not supported!");
}
@Override
- public void delPartition(String partitionName) throws IOException {
- throw new IOException("not supported!");
+ public void delPartition(String partitionName) throws CatalogException {
+ throw new RuntimeException("not supported!");
}
@Override
- public void delPartitions(String tableName) throws IOException {
- throw new IOException("not supported!");
+ public void delPartitions(String tableName) throws CatalogException {
+ throw new RuntimeException("not supported!");
}
/* (non-Javadoc)
* @see CatalogStore#addIndex(nta.catalog.proto.CatalogProtos.IndexDescProto)
*/
@Override
- public void addIndex(IndexDescProto proto) throws IOException {
+ public void addIndex(IndexDescProto proto) throws CatalogException {
synchronized(indexes) {
indexes.put(proto.getName(), proto);
indexesByColumn.put(proto.getTableId() + "."
@@ -179,7 +180,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#delIndex(java.lang.String)
*/
@Override
- public void delIndex(String indexName) throws IOException {
+ public void delIndex(String indexName) throws CatalogException {
synchronized(indexes) {
indexes.remove(indexName);
}
@@ -189,7 +190,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#getIndex(java.lang.String)
*/
@Override
- public IndexDescProto getIndex(String indexName) throws IOException {
+ public IndexDescProto getIndex(String indexName) throws CatalogException {
return indexes.get(indexName);
}
@@ -197,8 +198,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#getIndex(java.lang.String, java.lang.String)
*/
@Override
- public IndexDescProto getIndex(String tableName, String columnName)
- throws IOException {
+ public IndexDescProto getIndex(String tableName, String columnName) throws CatalogException {
return indexesByColumn.get(tableName+"."+columnName);
}
@@ -206,7 +206,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#existIndex(java.lang.String)
*/
@Override
- public boolean existIndex(String indexName) throws IOException {
+ public boolean existIndex(String indexName) throws CatalogException {
return indexes.containsKey(indexName);
}
@@ -214,8 +214,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#existIndex(java.lang.String, java.lang.String)
*/
@Override
- public boolean existIndex(String tableName, String columnName)
- throws IOException {
+ public boolean existIndex(String tableName, String columnName) throws CatalogException {
return indexesByColumn.containsKey(tableName + "." + columnName);
}
@@ -223,7 +222,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#getIndexes(java.lang.String)
*/
@Override
- public IndexDescProto[] getIndexes(String tableName) throws IOException {
+ public IndexDescProto[] getIndexes(String tableName) throws CatalogException {
List<IndexDescProto> protos = new ArrayList<IndexDescProto>();
for (IndexDescProto proto : indexesByColumn.values()) {
if (proto.getTableId().equals(tableName)) {
@@ -237,7 +236,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#addFunction(FunctionDesc)
*/
@Override
- public void addFunction(FunctionDesc func) throws IOException {
+ public void addFunction(FunctionDesc func) throws CatalogException {
// to be implemented
}
@@ -245,7 +244,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#deleteFunction(FunctionDesc)
*/
@Override
- public void deleteFunction(FunctionDesc func) throws IOException {
+ public void deleteFunction(FunctionDesc func) throws CatalogException {
// to be implemented
}
@@ -253,7 +252,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#existFunction(FunctionDesc)
*/
@Override
- public void existFunction(FunctionDesc func) throws IOException {
+ public void existFunction(FunctionDesc func) throws CatalogException {
// to be implemented
}
@@ -261,7 +260,7 @@ public class MemStore implements CatalogStore {
* @see CatalogStore#getAllFunctionNames()
*/
@Override
- public List<String> getAllFunctionNames() throws IOException {
+ public List<String> getAllFunctionNames() throws CatalogException {
// to be implemented
return null;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MySQLStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MySQLStore.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MySQLStore.java
index e9c5a03..d78e734 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MySQLStore.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MySQLStore.java
@@ -23,9 +23,9 @@ package org.apache.tajo.catalog.store;
import org.apache.hadoop.conf.Configuration;
import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.exception.CatalogException;
import org.apache.tajo.exception.InternalException;
-import java.io.IOException;
import java.sql.*;
import java.util.Map;
@@ -49,151 +49,185 @@ public class MySQLStore extends AbstractDBStore {
// TODO - DDL and index statements should be renamed
@Override
- protected void createBaseTable() throws IOException {
- int result;
+ protected void createBaseTable() throws CatalogException {
Statement stmt = null;
- Connection conn = getConnection();
+ Connection conn = null;
try {
+ conn = getConnection();
stmt = conn.createStatement();
+
+ StringBuilder sql = new StringBuilder();
// META
if (!baseTableMaps.get(TB_META)) {
- String meta_ddl = "CREATE TABLE " + TB_META + " (version int NOT NULL)";
+ sql.append("CREATE TABLE ").append(TB_META).append(" (version int NOT NULL)");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(meta_ddl);
+ LOG.debug(sql.toString());
}
- result = stmt.executeUpdate(meta_ddl);
+
+ stmt.executeUpdate(sql.toString());
LOG.info("Table '" + TB_META + " is created.");
}
// TABLES
if (!baseTableMaps.get(TB_TABLES)) {
- String tables_ddl = "CREATE TABLE "
- + TB_TABLES + " ("
- + "TID int NOT NULL AUTO_INCREMENT PRIMARY KEY, "
- + C_TABLE_ID + " VARCHAR(255) NOT NULL UNIQUE, "
- + "path TEXT, "
- + "store_type CHAR(16)"
- + ")";
+ sql.delete(0, sql.length());
+ sql.append("CREATE TABLE ").append(TB_TABLES).append("(");
+ sql.append("TID int NOT NULL AUTO_INCREMENT PRIMARY KEY, ");
+ sql.append(C_TABLE_ID).append(" VARCHAR(255) NOT NULL UNIQUE, ");
+ sql.append("path TEXT, ").append("store_type CHAR(16)").append(")");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(tables_ddl);
+ LOG.debug(sql.toString());
}
+ stmt.executeUpdate(sql.toString());
LOG.info("Table '" + TB_TABLES + "' is created.");
- result = stmt.executeUpdate(tables_ddl);
}
// COLUMNS
if (!baseTableMaps.get(TB_COLUMNS)) {
- String columns_ddl =
- "CREATE TABLE " + TB_COLUMNS + " ("
- + C_TABLE_ID + " VARCHAR(255) NOT NULL,"
- + "column_id INT NOT NULL,"
- + "column_name VARCHAR(255) NOT NULL, " + "data_type CHAR(16), " + "type_length INTEGER, "
- + "UNIQUE KEY(" + C_TABLE_ID + ", column_name),"
- + "FOREIGN KEY("+C_TABLE_ID+") REFERENCES "+TB_TABLES+"("+C_TABLE_ID+") ON DELETE CASCADE)";
+ sql.delete(0, sql.length());
+ sql.append("CREATE TABLE ").append(TB_COLUMNS).append("(");
+ sql.append(C_TABLE_ID).append(" VARCHAR(255) NOT NULL,");
+ sql.append("column_id INT NOT NULL,");
+ sql.append("column_name VARCHAR(255) NOT NULL, ");
+ sql.append("data_type CHAR(16), ");
+ sql.append("type_length INTEGER, ");
+ sql.append("UNIQUE KEY(").append(C_TABLE_ID).append(", column_name),");
+ sql.append("FOREIGN KEY(").append(C_TABLE_ID).append(") REFERENCES ");
+ sql.append(TB_TABLES).append("(").append(C_TABLE_ID).append(") ON DELETE CASCADE)");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(columns_ddl);
+ LOG.debug(sql.toString());
}
+ stmt.executeUpdate(sql.toString());
LOG.info("Table '" + TB_COLUMNS + " is created.");
- result = stmt.executeUpdate(columns_ddl);
}
// OPTIONS
if (!baseTableMaps.get(TB_OPTIONS)) {
- String options_ddl =
- "CREATE TABLE " + TB_OPTIONS + " ("
- + C_TABLE_ID + " VARCHAR(255) NOT NULL,"
- + "key_ VARCHAR(255) NOT NULL, value_ VARCHAR(255) NOT NULL,"
- + "INDEX("+C_TABLE_ID+", key_),"
- + "FOREIGN KEY("+C_TABLE_ID+") REFERENCES "+TB_TABLES+"("+C_TABLE_ID+") ON DELETE CASCADE)";
+ sql.delete(0, sql.length());
+ sql.append("CREATE TABLE ").append(TB_OPTIONS).append("(");
+ sql.append(C_TABLE_ID).append( " VARCHAR(255) NOT NULL,");
+ sql.append("key_ VARCHAR(255) NOT NULL, value_ VARCHAR(255) NOT NULL,");
+ sql.append("INDEX(").append(C_TABLE_ID).append(", key_),");
+ sql.append("FOREIGN KEY(").append(C_TABLE_ID);
+ sql.append(") REFERENCES ").append(TB_TABLES);
+ sql.append("(").append(C_TABLE_ID).append(") ON DELETE CASCADE)");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(options_ddl);
+ LOG.debug(sql.toString());
}
+
+ stmt.executeUpdate(sql.toString());
LOG.info("Table '" + TB_OPTIONS + " is created.");
- result = stmt.executeUpdate(options_ddl);
}
// INDEXES
if (!baseTableMaps.get(TB_INDEXES)) {
- String indexes_ddl = "CREATE TABLE " + TB_INDEXES + "("
- + "index_name VARCHAR(255) NOT NULL PRIMARY KEY, "
- + C_TABLE_ID + " VARCHAR(255) NOT NULL,"
- + "column_name VARCHAR(255) NOT NULL, "
- + "data_type VARCHAR(255) NOT NULL, "
- + "index_type CHAR(32) NOT NULL, "
- + "is_unique BOOLEAN NOT NULL, "
- + "is_clustered BOOLEAN NOT NULL, "
- + "is_ascending BOOLEAN NOT NULL,"
- + "INDEX(" + C_TABLE_ID + ", column_name),"
- + "FOREIGN KEY("+C_TABLE_ID+") REFERENCES "+TB_TABLES+"("+C_TABLE_ID+") ON DELETE CASCADE)";
+ sql.delete(0, sql.length());
+ sql.append("CREATE TABLE ").append(TB_INDEXES).append("(");
+ sql.append("index_name VARCHAR(255) NOT NULL PRIMARY KEY, ");
+ sql.append(C_TABLE_ID).append(" VARCHAR(255) NOT NULL,");
+ sql.append("column_name VARCHAR(255) NOT NULL, ");
+ sql.append("data_type VARCHAR(255) NOT NULL, ");
+ sql.append("index_type CHAR(32) NOT NULL, ");
+ sql.append("is_unique BOOLEAN NOT NULL, ");
+ sql.append("is_clustered BOOLEAN NOT NULL, ");
+ sql.append("is_ascending BOOLEAN NOT NULL,");
+ sql.append("INDEX(").append(C_TABLE_ID).append(", column_name),");
+ sql.append("FOREIGN KEY(").append(C_TABLE_ID);
+ sql.append(") REFERENCES ").append(TB_TABLES);
+ sql.append("(").append(C_TABLE_ID).append(") ON DELETE CASCADE)");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(indexes_ddl);
+ LOG.debug(sql.toString());
}
+
+ stmt.executeUpdate(sql.toString());
LOG.info("Table '" + TB_INDEXES + "' is created.");
- result = stmt.executeUpdate(indexes_ddl);
}
if (!baseTableMaps.get(TB_STATISTICS)) {
- String stats_ddl = "CREATE TABLE " + TB_STATISTICS + "("
- + C_TABLE_ID + " VARCHAR(255) NOT NULL,"
- + "num_rows BIGINT, "
- + "num_bytes BIGINT,"
- + "INDEX("+C_TABLE_ID+"),"
- + "FOREIGN KEY("+C_TABLE_ID+") REFERENCES "+TB_TABLES+"("+C_TABLE_ID+") ON DELETE CASCADE)";
+ sql.delete(0, sql.length());
+ sql.append("CREATE TABLE ").append(TB_STATISTICS).append("(");
+ sql.append(C_TABLE_ID).append(" VARCHAR(255) NOT NULL,");
+ sql.append("num_rows BIGINT, ");
+ sql.append("num_bytes BIGINT,");
+ sql.append("INDEX(").append(C_TABLE_ID).append("),");
+ sql.append("FOREIGN KEY(").append(C_TABLE_ID);
+ sql.append(") REFERENCES ").append(TB_TABLES);
+ sql.append("(").append(C_TABLE_ID).append(") ON DELETE CASCADE)");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(stats_ddl);
+ LOG.debug(sql.toString());
}
+
+ stmt.executeUpdate(sql.toString());
LOG.info("Table '" + TB_STATISTICS + "' is created.");
- result = stmt.executeUpdate(stats_ddl);
}
// PARTITION_METHODS
if (!baseTableMaps.get(TB_PARTITION_METHODS)) {
- String partition_method_ddl = "CREATE TABLE " + TB_PARTITION_METHODS + " ("
- + C_TABLE_ID + " VARCHAR(255) PRIMARY KEY,"
- + "partition_type VARCHAR(10) NOT NULL,"
- + "expression TEXT NOT NULL,"
- + "expression_schema VARBINARY(1024) NOT NULL, "
- + "FOREIGN KEY("+C_TABLE_ID+") REFERENCES "+TB_TABLES+"("+C_TABLE_ID+") ON DELETE CASCADE)";
+ sql.delete(0, sql.length());
+ sql.append("CREATE TABLE ").append(TB_PARTITION_METHODS).append("(");
+ sql.append(C_TABLE_ID).append(" VARCHAR(255) PRIMARY KEY,");
+ sql.append("partition_type VARCHAR(10) NOT NULL,");
+ sql.append("expression TEXT NOT NULL,");
+ sql.append("expression_schema VARBINARY(1024) NOT NULL, ");
+ sql.append("FOREIGN KEY(").append(C_TABLE_ID);
+ sql.append(") REFERENCES ").append(TB_TABLES);
+ sql.append("(").append(C_TABLE_ID).append(") ON DELETE CASCADE)");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(partition_method_ddl);
+ LOG.debug(sql.toString());
}
+
+ stmt.executeUpdate(sql.toString());
LOG.info("Table '" + TB_PARTITION_METHODS + "' is created.");
- result = stmt.executeUpdate(partition_method_ddl);
}
// PARTITIONS
if (!baseTableMaps.get(TB_PARTTIONS)) {
- String partition_ddl = "CREATE TABLE " + TB_PARTTIONS + " ("
- + "PID int NOT NULL AUTO_INCREMENT PRIMARY KEY, "
- + C_TABLE_ID + " VARCHAR(255) NOT NULL,"
- + "partition_name VARCHAR(255), "
- + "ordinal_position INT NOT NULL,"
- + "partition_value TEXT,"
- + "path TEXT,"
- + "cache_nodes VARCHAR(255), "
- + "UNIQUE KEY(" + C_TABLE_ID + ", partition_name),"
- + "FOREIGN KEY("+C_TABLE_ID+") REFERENCES "+TB_TABLES+"("+C_TABLE_ID+") ON DELETE CASCADE)";
+ sql.delete(0, sql.length());
+ sql.append("CREATE TABLE ").append(TB_PARTTIONS).append("(");
+ sql.append("PID int NOT NULL AUTO_INCREMENT PRIMARY KEY, ");
+ sql.append(C_TABLE_ID).append( " VARCHAR(255) NOT NULL,");
+ sql.append("partition_name VARCHAR(255), ");
+ sql.append("ordinal_position INT NOT NULL,");
+ sql.append("partition_value TEXT,");
+ sql.append("path TEXT,");
+ sql.append("cache_nodes VARCHAR(255), ");
+ sql.append("UNIQUE KEY(").append(C_TABLE_ID).append(", partition_name),");
+ sql.append("FOREIGN KEY(").append(C_TABLE_ID);
+ sql.append(") REFERENCES ").append(TB_TABLES);
+ sql.append("(").append(C_TABLE_ID).append(") ON DELETE CASCADE)");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(partition_ddl);
+ LOG.debug(sql.toString());
}
+
+ stmt.executeUpdate(sql.toString());
+
LOG.info("Table '" + TB_PARTTIONS + "' is created.");
- result = stmt.executeUpdate(partition_ddl);
}
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(stmt);
+ CatalogUtil.closeQuietly(conn, stmt);
}
}
@Override
- protected boolean isInitialized() throws IOException {
+ protected boolean isInitialized() throws CatalogException {
+ Connection conn = null;
ResultSet res = null;
- Connection conn = getConnection();
+
try {
+ conn = getConnection();
res = conn.getMetaData().getTables(null, null, null,
new String[]{"TABLE"});
@@ -212,20 +246,20 @@ public class MySQLStore extends AbstractDBStore {
while (res.next()) {
baseTableMaps.put(res.getString("TABLE_NAME"), true);
}
- } catch(SQLException se) {
- throw new IOException(se);
- } finally {
- CatalogUtil.closeSQLWrapper(res);
- }
- for(Map.Entry<String, Boolean> entry : baseTableMaps.entrySet()) {
- if (!entry.getValue()) {
- return false;
+ for(Map.Entry<String, Boolean> entry : baseTableMaps.entrySet()) {
+ if (!entry.getValue()) {
+ return false;
+ }
}
+
+ } catch(SQLException se) {
+ throw new CatalogException(se);
+ } finally {
+ CatalogUtil.closeQuietly(conn, res);
}
return true;
-// return false;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/global/DataChannel.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/global/DataChannel.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/global/DataChannel.java
index efa1e05..9f5c6bf 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/global/DataChannel.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/global/DataChannel.java
@@ -22,6 +22,7 @@ import com.google.common.base.Preconditions;
import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.catalog.Column;
import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.engine.utils.SchemaUtil;
import static org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
import static org.apache.tajo.ipc.TajoWorkerProtocol.*;
@@ -175,7 +176,7 @@ public class DataChannel {
}
public void setSchema(Schema schema) {
- this.schema = (Schema) schema.clone();
+ this.schema = SchemaUtil.clone(schema);
}
public Schema getSchema() {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/TableSubQueryNode.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/TableSubQueryNode.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/TableSubQueryNode.java
index 7b88fa5..e205454 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/TableSubQueryNode.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/TableSubQueryNode.java
@@ -24,6 +24,7 @@ import org.apache.tajo.catalog.Schema;
import org.apache.tajo.engine.planner.PlanString;
import org.apache.tajo.engine.planner.PlannerUtil;
import org.apache.tajo.engine.planner.Target;
+import org.apache.tajo.engine.utils.SchemaUtil;
import org.apache.tajo.util.TUtil;
public class TableSubQueryNode extends RelationNode implements Projectable {
@@ -36,8 +37,8 @@ public class TableSubQueryNode extends RelationNode implements Projectable {
this.tableName = PlannerUtil.normalizeTableName(tableName);
if (subQuery != null) {
this.subQuery = subQuery;
- setOutSchema((Schema) this.subQuery.getOutSchema().clone());
- setInSchema((Schema) this.subQuery.getOutSchema().clone());
+ setOutSchema(SchemaUtil.clone(this.subQuery.getOutSchema()));
+ setInSchema(SchemaUtil.clone(this.subQuery.getOutSchema()));
getInSchema().setQualifier(this.tableName);
getOutSchema().setQualifier(this.tableName);
}
@@ -66,12 +67,12 @@ public class TableSubQueryNode extends RelationNode implements Projectable {
public void setSubQuery(LogicalNode node) {
this.subQuery = node;
- setInSchema((Schema) this.subQuery.getOutSchema().clone());
+ setInSchema(SchemaUtil.clone(this.subQuery.getOutSchema()));
getInSchema().setQualifier(this.tableName);
if (hasTargets()) {
setOutSchema(PlannerUtil.targetToSchema(targets));
} else {
- setOutSchema((Schema) this.subQuery.getOutSchema().clone());
+ setOutSchema(SchemaUtil.clone(this.subQuery.getOutSchema()));
}
getOutSchema().setQualifier(this.tableName);
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
index a0c0eeb..53a1c24 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
@@ -31,6 +31,7 @@ import org.apache.tajo.engine.eval.FieldEval;
import org.apache.tajo.engine.planner.Projector;
import org.apache.tajo.engine.planner.Target;
import org.apache.tajo.engine.planner.logical.ScanNode;
+import org.apache.tajo.engine.utils.SchemaUtil;
import org.apache.tajo.engine.utils.TupleUtil;
import org.apache.tajo.storage.*;
import org.apache.tajo.storage.fragment.FileFragment;
@@ -73,7 +74,7 @@ public class SeqScanExec extends PhysicalExec {
*/
private void rewriteColumnPartitionedTableSchema() throws IOException {
PartitionMethodDesc partitionDesc = plan.getTableDesc().getPartitionMethod();
- Schema columnPartitionSchema = (Schema) partitionDesc.getExpressionSchema().clone();
+ Schema columnPartitionSchema = SchemaUtil.clone(partitionDesc.getExpressionSchema());
String qualifier = inSchema.getColumn(0).getQualifier();
columnPartitionSchema.setQualifier(qualifier);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/utils/SchemaUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/utils/SchemaUtil.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/utils/SchemaUtil.java
index 9f7d23d..c9ab823 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/utils/SchemaUtil.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/utils/SchemaUtil.java
@@ -60,4 +60,13 @@ public class SchemaUtil {
}
return logicalSchema;
}
+
+ public static <T extends Schema> T clone(Schema schema) {
+ try {
+ T copy = (T) schema.clone();
+ return copy;
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeException(e);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java
index fbfc76f..dea4fc8 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java
@@ -30,6 +30,7 @@ import org.apache.tajo.engine.planner.LogicalPlan;
import org.apache.tajo.engine.planner.LogicalPlanner;
import org.apache.tajo.engine.planner.PlanningException;
import org.apache.tajo.engine.planner.Target;
+import org.apache.tajo.engine.utils.SchemaUtil;
import org.apache.tajo.master.TajoMaster;
import org.apache.tajo.storage.LazyTuple;
import org.apache.tajo.storage.Tuple;
@@ -102,7 +103,7 @@ public class ExprTestBase {
VTuple vtuple = null;
Schema inputSchema = null;
if (schema != null) {
- inputSchema = (Schema) schema.clone();
+ inputSchema = SchemaUtil.clone(schema);
inputSchema.setQualifier(tableName);
int targetIdx [] = new int[inputSchema.getColumnNum()];
[2/3] TAJO-530: Fix warnings in tajo-catalog. (jaehwa)
Posted by bl...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
index 1ac9d80..c0eecf2 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
@@ -28,31 +28,31 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.tajo.catalog.CatalogConstants;
import org.apache.tajo.catalog.CatalogUtil;
import org.apache.tajo.catalog.FunctionDesc;
+import org.apache.tajo.catalog.exception.CatalogException;
import org.apache.tajo.catalog.proto.CatalogProtos;
-import org.apache.tajo.catalog.proto.CatalogProtos.ColumnProto;
-import org.apache.tajo.catalog.proto.CatalogProtos.IndexDescProto;
-import org.apache.tajo.catalog.proto.CatalogProtos.IndexMethod;
-import org.apache.tajo.catalog.proto.CatalogProtos.SchemaProto;
-import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
-import org.apache.tajo.catalog.proto.CatalogProtos.TableStatsProto;
-
+import org.apache.tajo.catalog.proto.CatalogProtos.*;
import org.apache.tajo.common.TajoDataTypes.Type;
import org.apache.tajo.exception.InternalException;
-import java.io.IOException;
-import java.sql.*;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
import java.util.ArrayList;
-import java.util.Map;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
public abstract class AbstractDBStore extends CatalogConstants implements CatalogStore {
protected final Log LOG = LogFactory.getLog(getClass());
+
protected Configuration conf;
protected String connectionId;
protected String connectionPassword;
protected String catalogUri;
+
private Connection conn;
+
protected Map<String, Boolean> baseTableMaps = new HashMap<String, Boolean>();
protected static final int VERSION = 1;
@@ -61,9 +61,9 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
protected abstract Connection createConnection(final Configuration conf) throws SQLException;
- protected abstract boolean isInitialized() throws IOException;
+ protected abstract boolean isInitialized() throws CatalogException;
- protected abstract void createBaseTable() throws IOException;
+ protected abstract void createBaseTable() throws CatalogException;
public AbstractDBStore(Configuration conf)
throws InternalException {
@@ -99,7 +99,7 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
Class.forName(getCatalogDriverName()).newInstance();
LOG.info("Loaded the Catalog driver (" + catalogDriver + ")");
} catch (Exception e) {
- throw new InternalException("Cannot load Catalog driver " + catalogDriver, e);
+ throw new CatalogException("Cannot load Catalog driver " + catalogDriver, e);
}
try {
@@ -107,7 +107,7 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
conn = createConnection(conf);
LOG.info("Connected to database (" + catalogUri + ")");
} catch (SQLException e) {
- throw new InternalException("Cannot connect to database (" + catalogUri
+ throw new CatalogException("Cannot connect to database (" + catalogUri
+ ")", e);
}
@@ -118,17 +118,18 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
} else {
LOG.info("The base tables of CatalogServer already is initialized.");
}
- } catch (IOException se) {
- throw new InternalException(
+ } catch (Exception se) {
+ throw new CatalogException(
"Cannot initialize the persistent storage of Catalog", se);
}
- int dbVersion = 0;
+// int dbVersion = 0;
try {
- dbVersion = needUpgrade();
- } catch (IOException e) {
- throw new InternalException(
- "Cannot check if the DB need to be upgraded", e);
+// dbVersion = needUpgrade();
+ needUpgrade();
+ } catch (Exception e) {
+ throw new CatalogException(
+ "Cannot check if the DB need to be upgraded.", e);
}
// if (dbVersion < VERSION) {
@@ -146,69 +147,106 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
return catalogUri;
}
- public Connection getConnection() throws IOException {
+ public Connection getConnection() {
try {
boolean isValid = conn.isValid(100);
if (!isValid) {
- CatalogUtil.closeSQLWrapper(conn);
+ CatalogUtil.closeQuietly(conn);
conn = createConnection(conf);
}
} catch (SQLException e) {
- CatalogUtil.closeSQLWrapper(conn);
- throw new IOException(e);
+ e.printStackTrace();
}
return conn;
}
- private int needUpgrade() throws IOException {
- Statement stmt = null;
+ private int needUpgrade() throws CatalogException {
+ Connection conn = null;
+ PreparedStatement pstmt = null;
+ ResultSet res = null;
+ int retValue = -1;
+
try {
- String sql = "SELECT VERSION FROM " + TB_META;
- stmt = getConnection().createStatement();
- ResultSet res = stmt.executeQuery(sql);
+ StringBuilder sql = new StringBuilder();
+ sql.delete(0, sql.length());
+ sql.append("SELECT VERSION FROM ");
+ sql.append(TB_META);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ res = pstmt.executeQuery();
if (!res.next()) { // if this db version is 0
insertVersion();
- return 0;
+ retValue = 0;
} else {
- return res.getInt(1);
+ retValue = res.getInt(1);
}
} catch (SQLException e) {
- throw new IOException(e);
+ throw new CatalogException(e);
} finally {
- CatalogUtil.closeSQLWrapper(stmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
+ return retValue;
}
- private void insertVersion() throws IOException {
- Statement stmt = null;
+ private void insertVersion() throws CatalogException {
+ Connection conn = null;
+ PreparedStatement pstmt = null;
+
try {
- String sql = "INSERT INTO " + TB_META + " values (0)";
- stmt = getConnection().createStatement();
- stmt.executeUpdate(sql);
+ StringBuilder sql = new StringBuilder();
+ sql.append("INSERT INTO ");
+ sql.append(TB_META);
+ sql.append(" values (0)");
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.executeUpdate();
} catch (SQLException e) {
- throw new IOException(e);
+ throw new CatalogException(e);
} finally {
- CatalogUtil.closeSQLWrapper(stmt);
+ CatalogUtil.closeQuietly(conn, pstmt);
}
}
- private void upgrade(int from, int to) throws IOException {
- Statement stmt = null;
+ private void upgrade(int from, int to) throws CatalogException {
+ Connection conn = null;
+ PreparedStatement pstmt = null;
+
try {
if (from == 0) {
if (to == 1) {
- String sql = "DROP INDEX idx_options_key";
- LOG.info(sql);
+ StringBuilder sql = new StringBuilder();
+ sql.append(" DROP INDEX idx_options_key");
- stmt = getConnection().createStatement();
- stmt.addBatch(sql);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.executeUpdate();
+ pstmt.close();
+
+ sql.delete(0, sql.length());
+ sql.append(" CREATE INDEX idx_options_key on ").append(TB_OPTIONS);
+ sql.append(" (").append(C_TABLE_ID).append(")");
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
- sql =
- "CREATE INDEX idx_options_key on " + TB_OPTIONS + " (" + C_TABLE_ID + ")";
- stmt.addBatch(sql);
- LOG.info(sql);
- stmt.executeBatch();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.executeUpdate();
LOG.info("DB Upgraded from " + from + " to " + to);
} else {
@@ -216,48 +254,57 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
}
}
} catch (SQLException e) {
- throw new IOException(e);
+ throw new CatalogException(e);
} finally {
- CatalogUtil.closeSQLWrapper(stmt);
+ CatalogUtil.closeQuietly(conn, pstmt);
}
}
@Override
- public void addTable(final CatalogProtos.TableDescProto table) throws IOException {
+ public void addTable(final CatalogProtos.TableDescProto table) throws CatalogException {
+ Connection conn = null;
PreparedStatement pstmt = null;
ResultSet res = null;
- Connection conn = getConnection();
+
try {
+ conn = getConnection();
conn.setAutoCommit(false);
-
String tableName = table.getId().toLowerCase();
String sql = String.format("INSERT INTO %s (%s, path, store_type) VALUES(?, ?, ?) ", TB_TABLES, C_TABLE_ID);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql);
+ }
+
pstmt = conn.prepareStatement(sql);
pstmt.setString(1, tableName);
pstmt.setString(2, table.getPath());
pstmt.setString(3, table.getMeta().getStoreType().name());
- if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
- }
pstmt.executeUpdate();
+ pstmt.close();
String tidSql = String.format("SELECT TID from %s WHERE %s = ?", TB_TABLES, C_TABLE_ID);
pstmt = conn.prepareStatement(tidSql);
pstmt.setString(1, tableName);
- if (LOG.isDebugEnabled()) {
- LOG.debug(tidSql);
- }
res = pstmt.executeQuery();
+
if (!res.next()) {
- throw new IOException("ERROR: there is no tid matched to "
- + table.getId());
+ throw new CatalogException("ERROR: there is no tid matched to " + table.getId());
}
+
int tid = res.getInt("TID");
+ res.close();
+ pstmt.close();
String colSql = String.format("INSERT INTO %s (TID, %s, column_id, column_name, data_type, type_length)"
+ " VALUES(?, ?, ?, ?, ?, ?) ", TB_COLUMNS, C_TABLE_ID);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql);
+ }
+
pstmt = conn.prepareStatement(colSql);
for(int i = 0; i < table.getSchema().getFieldsCount(); i++) {
ColumnProto col = table.getSchema().getFields(i);
@@ -268,46 +315,56 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
pstmt.setString(5, col.getDataType().getType().name());
pstmt.setInt(6, (col.getDataType().hasLength() ? col.getDataType().getLength() : 0));
pstmt.addBatch();
- if (LOG.isDebugEnabled()) {
- LOG.debug(colSql);
- }
+ pstmt.clearParameters();
}
pstmt.executeBatch();
+ pstmt.close();
if(table.getMeta().hasParams()) {
String optSql = String.format("INSERT INTO %s (%s, key_, value_) VALUES(?, ?, ?)", TB_OPTIONS, C_TABLE_ID);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(optSql);
+ }
+
pstmt = conn.prepareStatement(optSql);
for (CatalogProtos.KeyValueProto entry : table.getMeta().getParams().getKeyvalList()) {
pstmt.setString(1, tableName);
pstmt.setString(2, entry.getKey());
pstmt.setString(3, entry.getValue());
pstmt.addBatch();
- if (LOG.isDebugEnabled()) {
- LOG.debug(optSql);
- }
+ pstmt.clearParameters();
}
pstmt.executeBatch();
+ pstmt.close();
}
-
if (table.hasStats()) {
+
String statSql =
String.format("INSERT INTO %s (%s, num_rows, num_bytes) VALUES(?, ?, ?)", TB_STATISTICS, C_TABLE_ID);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(statSql);
+ }
+
pstmt = conn.prepareStatement(statSql);
pstmt.setString(1, tableName);
pstmt.setLong(2, table.getStats().getNumRows());
pstmt.setLong(3, table.getStats().getNumBytes());
- pstmt.addBatch();
- if (LOG.isDebugEnabled()) {
- LOG.debug(statSql);
- }
- pstmt.executeBatch();
+ pstmt.executeUpdate();
+ pstmt.close();
}
if(table.hasPartition()) {
String partSql =
String.format("INSERT INTO %s (%s, partition_type, expression, expression_schema) VALUES(?, ?, ?, ?)",
TB_PARTITION_METHODS, C_TABLE_ID);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(partSql);
+ }
+
pstmt = conn.prepareStatement(partSql);
pstmt.setString(1, tableName);
pstmt.setString(2, table.getPartition().getPartitionType().name());
@@ -316,199 +373,233 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
pstmt.executeUpdate();
}
+ // If there is no error, commit the changes.
conn.commit();
} catch (SQLException se) {
try {
+ // If there is any error, rollback the changes.
conn.rollback();
- } catch (SQLException rbe) {
- throw new IOException(se.getMessage(), rbe);
+ } catch (SQLException se2) {
}
- throw new IOException(se.getMessage(), se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(pstmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
}
@Override
- public boolean existTable(final String name) throws IOException {
- StringBuilder sql = new StringBuilder();
- sql.append("SELECT " + C_TABLE_ID + " from ")
- .append(TB_TABLES)
- .append(" WHERE " + C_TABLE_ID + " = '")
- .append(name)
- .append("'");
-
- Statement stmt = null;
+ public boolean existTable(final String name) throws CatalogException {
+ Connection conn = null;
+ PreparedStatement pstmt = null;
+ ResultSet res = null;
boolean exist = false;
try {
- stmt = getConnection().createStatement();
+ StringBuilder sql = new StringBuilder();
+ sql.append(" SELECT ").append(C_TABLE_ID);
+ sql.append(" FROM ").append(TB_TABLES);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
+
if (LOG.isDebugEnabled()) {
LOG.debug(sql.toString());
}
- ResultSet res = stmt.executeQuery(sql.toString());
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+
+ pstmt.setString(1, name);
+ res = pstmt.executeQuery();
exist = res.next();
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(stmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
return exist;
}
@Override
- public void deleteTable(final String name) throws IOException {
- Statement stmt = null;
- String sql = null;
+ public void deleteTable(final String name) throws CatalogException {
+ Connection conn = null;
+ PreparedStatement pstmt = null;
try {
- getConnection().setAutoCommit(false);
+ conn = getConnection();
+ conn.setAutoCommit(false);
- } catch (SQLException se) {
- throw new IOException(se);
- } finally {
- CatalogUtil.closeSQLWrapper(stmt);
- }
- try {
- stmt = getConnection().createStatement();
- sql = "DELETE FROM " + TB_COLUMNS +
- " WHERE " + C_TABLE_ID + " = '" + name + "'";
- LOG.info(sql);
- stmt.execute(sql);
- } catch (SQLException se) {
- throw new IOException(se);
- } finally {
- CatalogUtil.closeSQLWrapper(stmt);
- }
+ StringBuilder sql = new StringBuilder();
+ sql.append("DELETE FROM ").append(TB_COLUMNS);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
- try {
- sql = "DELETE FROM " + TB_OPTIONS +
- " WHERE " + C_TABLE_ID + " = '" + name + "'";
- LOG.info(sql);
- stmt = getConnection().createStatement();
- stmt.execute(sql);
- } catch (SQLException se) {
- throw new IOException(se);
- } finally {
- CatalogUtil.closeSQLWrapper(stmt);
- }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
- try {
- sql = "DELETE FROM " + TB_STATISTICS +
- " WHERE " + C_TABLE_ID + " = '" + name + "'";
- LOG.info(sql);
- stmt = getConnection().createStatement();
- stmt.execute(sql);
- } catch (SQLException se) {
- throw new IOException(se);
- } finally {
- CatalogUtil.closeSQLWrapper(stmt);
- }
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, name);
+ pstmt.executeUpdate();
+ pstmt.close();
+ sql.delete(0, sql.length());
+ sql.append("DELETE FROM ").append(TB_OPTIONS);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
- try {
- sql = "DELETE FROM " + TB_PARTTIONS +
- " WHERE " + C_TABLE_ID + " = '" + name + "'";
- LOG.info(sql);
- stmt = getConnection().createStatement();
- stmt.execute(sql);
- } catch (SQLException se) {
- throw new IOException(se);
- } finally {
- CatalogUtil.closeSQLWrapper(stmt);
- }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
- try {
- sql = "DELETE FROM " + TB_TABLES +
- " WHERE " + C_TABLE_ID + " = '" + name + "'";
- LOG.info(sql);
- stmt = getConnection().createStatement();
- stmt.execute(sql);
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, name);
+ pstmt.executeUpdate();
+ pstmt.close();
+
+ sql.delete(0, sql.length());
+ sql.append("DELETE FROM ").append(TB_STATISTICS);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
+
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, name);
+ pstmt.executeUpdate();
+ pstmt.close();
+
+ sql.delete(0, sql.length());
+ sql.append("DELETE FROM ").append(TB_PARTTIONS);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
+
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, name);
+ pstmt.executeUpdate();
+ pstmt.close();
+
+ sql.delete(0, sql.length());
+ sql.append("DELETE FROM ").append(TB_TABLES);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
+
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, name);
+ pstmt.executeUpdate();
+
+ // If there is no error, commit the changes.
+ conn.commit();
} catch (SQLException se) {
- throw new IOException(se);
+ try {
+ // If there is any error, rollback the changes.
+ conn.rollback();
+ } catch (SQLException se2) {
+ }
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(stmt);
+ CatalogUtil.closeQuietly(conn, pstmt);
}
-
}
@Override
- public CatalogProtos.TableDescProto getTable(final String name) throws IOException {
+ public CatalogProtos.TableDescProto getTable(final String name) throws CatalogException {
+ Connection conn = null;
ResultSet res = null;
- Statement stmt = null;
+ PreparedStatement pstmt = null;
- CatalogProtos.TableDescProto.Builder tableBuilder = CatalogProtos.TableDescProto.newBuilder();
+ CatalogProtos.TableDescProto.Builder tableBuilder = null;
StoreType storeType = null;
+
try {
- String sql =
- "SELECT " + C_TABLE_ID + ", path, store_type from " + TB_TABLES
- + " WHERE " + C_TABLE_ID + "='" + name + "'";
+ tableBuilder = CatalogProtos.TableDescProto.newBuilder();
+
+ StringBuilder sql = new StringBuilder();
+ sql.append(" SELECT ").append(C_TABLE_ID).append(", path, store_type");
+ sql.append(" from ").append(TB_TABLES);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- stmt = getConnection().createStatement();
- res = stmt.executeQuery(sql);
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, name);
+ res = pstmt.executeQuery();
+
if (!res.next()) { // there is no table of the given name.
return null;
}
+
tableBuilder.setId(res.getString(C_TABLE_ID).trim());
tableBuilder.setPath(res.getString("path").trim());
-
storeType = CatalogUtil.getStoreType(res.getString("store_type").trim());
- } catch (SQLException se) {
- throw new IOException(se);
- } finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
- }
+ res.close();
+ pstmt.close();
- CatalogProtos.SchemaProto.Builder schemaBuilder = CatalogProtos.SchemaProto.newBuilder();
- try {
- String sql = "SELECT column_name, data_type, type_length from " + TB_COLUMNS
- + " WHERE " + C_TABLE_ID + "='" + name + "' ORDER by column_id asc";
+ CatalogProtos.SchemaProto.Builder schemaBuilder = CatalogProtos.SchemaProto.newBuilder();
+ sql.delete(0, sql.length());
+ sql.append(" SELECT column_name, data_type, type_length ");
+ sql.append(" from ").append(TB_COLUMNS);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
+ sql.append("ORDER by column_id asc");
- stmt = getConnection().createStatement();
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- res = stmt.executeQuery(sql);
+
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, name);
+ res = pstmt.executeQuery();
+
while (res.next()) {
schemaBuilder.addFields(resultToColumnProto(res));
}
- } catch (SQLException se) {
- throw new IOException(se);
- } finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
- }
- tableBuilder.setSchema(CatalogUtil.getQualfiedSchema(name, schemaBuilder.build()));
- CatalogProtos.TableProto.Builder metaBuilder = CatalogProtos.TableProto.newBuilder();
- metaBuilder.setStoreType(storeType);
- try {
- String sql = "SELECT key_, value_ from " + TB_OPTIONS
- + " WHERE " + C_TABLE_ID + "='" + name + "'";
- stmt = getConnection().createStatement();
+ tableBuilder.setSchema(CatalogUtil.getQualfiedSchema(name, schemaBuilder.build()));
+
+ res.close();
+ pstmt.close();
+
+ CatalogProtos.TableProto.Builder metaBuilder = CatalogProtos.TableProto.newBuilder();
+ metaBuilder.setStoreType(storeType);
+ sql.delete(0, sql.length());
+ sql.append(" SELECT key_, value_ ");
+ sql.append(" from ").append(TB_OPTIONS);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- res = stmt.executeQuery(sql);
+
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, name);
+ res = pstmt.executeQuery();
metaBuilder.setParams(resultToKeyValueSetProto(res));
- } catch (SQLException se) {
- throw new IOException(se);
- } finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
- }
- tableBuilder.setMeta(metaBuilder);
- try {
- String sql = "SELECT num_rows, num_bytes from " + TB_STATISTICS
- + " WHERE " + C_TABLE_ID + "='" + name + "'";
+ tableBuilder.setMeta(metaBuilder);
+
+ res.close();
+ pstmt.close();
+
+ sql.delete(0, sql.length());
+ sql.append(" SELECT num_rows, num_bytes ");
+ sql.append(" from ").append(TB_STATISTICS);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- stmt = getConnection().createStatement();
- res = stmt.executeQuery(sql);
+
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, name);
+ res = pstmt.executeQuery();
if (res.next()) {
TableStatsProto.Builder statBuilder = TableStatsProto.newBuilder();
@@ -516,54 +607,66 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
statBuilder.setNumBytes(res.getLong("num_bytes"));
tableBuilder.setStats(statBuilder);
}
- } catch (SQLException se) {
- throw new IOException(se);
- } finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
- }
- try {
- String sql =
- "SELECT partition_type, expression, expression_schema from " + TB_PARTITION_METHODS
- + " WHERE " + C_TABLE_ID + "='" + name + "'";
+ res.close();
+ pstmt.close();
+
+ sql.delete(0, sql.length());
+ sql.append(" SELECT partition_type, expression, expression_schema ");
+ sql.append(" from ").append(TB_PARTITION_METHODS);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- stmt = getConnection().createStatement();
- res = stmt.executeQuery(sql);
+
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, name);
+ res = pstmt.executeQuery();
if (res.next()) {
tableBuilder.setPartition(resultToPartitionMethodProto(name, res));
}
+ } catch (InvalidProtocolBufferException e) {
+ throw new CatalogException(e);
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
return tableBuilder.build();
}
- private ColumnProto getColumn(String tableName, int tid, String columnId) throws IOException {
+ private ColumnProto getColumn(String tableName, int tid, String columnId) throws CatalogException {
+ Connection conn = null;
ResultSet res = null;
- Statement stmt = null;
+ PreparedStatement pstmt = null;
+
try {
- String sql = "SELECT column_name, data_type, type_length from "
- + TB_COLUMNS + " WHERE TID = " + tid + " AND column_id = " + columnId;
+ StringBuilder sql = new StringBuilder();
+ sql.append(" SELECT column_name, data_type, type_length");
+ sql.append(" FROM ").append(TB_COLUMNS);
+ sql.append(" WHERE TID = ? ");
+ sql.append(" AND column_id = ? ");
- stmt = getConnection().createStatement();
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- res = stmt.executeQuery(sql);
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setInt(1, tid);
+ pstmt.setInt(2, Integer.parseInt(columnId));
+ res = pstmt.executeQuery();
if (res.next()) {
return resultToColumnProto(res);
}
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
return null;
}
@@ -578,42 +681,59 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
}
@Override
- public List<String> getAllTableNames() throws IOException {
- String sql = "SELECT " + C_TABLE_ID + " from " + TB_TABLES;
-
- Statement stmt = null;
+ public List<String> getAllTableNames() throws CatalogException {
+ Connection conn = null;
+ PreparedStatement pstmt = null;
ResultSet res = null;
List<String> tables = new ArrayList<String>();
try {
- stmt = getConnection().createStatement();
+ StringBuilder sql = new StringBuilder();
+ sql.append("SELECT ");
+ sql.append(C_TABLE_ID);
+ sql.append(" from ");
+ sql.append(TB_TABLES);
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- res = stmt.executeQuery(sql);
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ res = pstmt.executeQuery();
while (res.next()) {
tables.add(res.getString(C_TABLE_ID).trim());
}
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
return tables;
}
@Override
- public void addPartitions(CatalogProtos.PartitionsProto partitionsProto) throws IOException {
+ public void addPartitions(CatalogProtos.PartitionsProto partitionsProto) throws CatalogException {
+ Connection conn = null;
PreparedStatement pstmt = null;
- String sql = "INSERT INTO " + TB_PARTTIONS + " (" + C_TABLE_ID
- + ", partition_name, ordinal_position, path, cache_nodes) "
- + "VALUES (?, ?, ?, ?, ?, ?) ";
- if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
- }
+
try {
- pstmt = getConnection().prepareStatement(sql);
+ StringBuilder sql = new StringBuilder();
+ sql.append("INSERT INTO ");
+ sql.append(TB_PARTTIONS);
+ sql.append(" (");
+ sql.append(C_TABLE_ID);
+ sql.append(", partition_name, ordinal_position, path, cache_nodes) ");
+ sql.append("VALUES (?, ?, ?, ?, ?, ?) ");
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+
for (CatalogProtos.PartitionDescProto proto : partitionsProto.getPartitionList()) {
pstmt.setString(1, proto.getTableId());
pstmt.setString(2, proto.getPartitionName());
@@ -621,123 +741,158 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
pstmt.setString(4, proto.getPartitionValue());
pstmt.setString(5, proto.getPath());
pstmt.addBatch();
+ pstmt.clearParameters();
}
pstmt.executeBatch();
} catch (SQLException se) {
- throw new IOException(se.getMessage(), se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(pstmt);
+ CatalogUtil.closeQuietly(conn, pstmt);
}
}
@Override
- public void addPartitionMethod(CatalogProtos.PartitionMethodProto proto) throws IOException {
+ public void addPartitionMethod(CatalogProtos.PartitionMethodProto proto) throws CatalogException {
+ Connection conn = null;
PreparedStatement pstmt = null;
- String sql = "INSERT INTO " + TB_PARTITION_METHODS + " (" + C_TABLE_ID
- + ", partition_type, expression, expression_schema) "
- + "VALUES (?, ?, ?, ?) ";
- if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
- }
try {
- pstmt = getConnection().prepareStatement(sql);
+ StringBuilder sql = new StringBuilder();
+ sql.append("INSERT INTO ");
+ sql.append(TB_PARTITION_METHODS);
+ sql.append( " (");
+ sql.append(C_TABLE_ID);
+ sql.append(", partition_type, expression, expression_schema) ");
+ sql.append("VALUES (?, ?, ?, ?) ");
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
pstmt.setString(1, proto.getTableId().toLowerCase());
pstmt.setString(2, proto.getPartitionType().name());
pstmt.setString(3, proto.getExpression());
pstmt.setBytes(4, proto.getExpressionSchema().toByteArray());
pstmt.executeUpdate();
} catch (SQLException se) {
- throw new IOException(se.getMessage(), se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(pstmt);
+ CatalogUtil.closeQuietly(conn, pstmt);
}
}
@Override
- public void delPartitionMethod(String tableName) throws IOException {
- String sql =
- "DELETE FROM " + TB_PARTITION_METHODS
- + " WHERE " + C_TABLE_ID + " ='" + tableName + "'";
- if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
- }
- Statement stmt = null;
+ public void delPartitionMethod(String tableName) throws CatalogException {
+ Connection conn = null;
+ PreparedStatement pstmt = null;
try {
- stmt = getConnection().createStatement();
+ StringBuilder sql = new StringBuilder();
+ sql.append(" DELETE FROM ").append(TB_PARTITION_METHODS);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- stmt.executeUpdate(sql);
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, tableName);
+ pstmt.executeUpdate();
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(stmt);
+ CatalogUtil.closeQuietly(conn, pstmt);
}
}
@Override
- public CatalogProtos.PartitionMethodProto getPartitionMethod(String tableName) throws IOException {
+ public CatalogProtos.PartitionMethodProto getPartitionMethod(String tableName) throws CatalogException {
+ Connection conn = null;
ResultSet res = null;
- Statement stmt = null;
+ PreparedStatement pstmt = null;
try {
- String sql =
- "SELECT partition_type, expression, expression_schema from " + TB_PARTITION_METHODS
- + " WHERE " + C_TABLE_ID + "='" + tableName + "'";
+ StringBuilder sql = new StringBuilder();
+ sql.append(" SELECT partition_type, expression, expression_schema");
+ sql.append(" FROM ").append(TB_PARTITION_METHODS);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- stmt = getConnection().createStatement();
- res = stmt.executeQuery(sql);
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, tableName);
+ res = pstmt.executeQuery();
if (res.next()) {
return resultToPartitionMethodProto(tableName, res);
}
+ } catch (InvalidProtocolBufferException e) {
+ throw new CatalogException(e);
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
return null;
}
@Override
- public boolean existPartitionMethod(String tableName) throws IOException {
+ public boolean existPartitionMethod(String tableName) throws CatalogException {
+ Connection conn = null;
ResultSet res = null;
- Statement stmt = null;
+ PreparedStatement pstmt = null;
boolean exist = false;
+
try {
- String sql =
- "SELECT partition_type, expression, expression_schema from " + TB_PARTITION_METHODS
- + " WHERE " + C_TABLE_ID + "='" + tableName + "'";
+ StringBuilder sql = new StringBuilder();
+ sql.append(" SELECT partition_type, expression, expression_schema");
+ sql.append(" FROM ").append(TB_PARTITION_METHODS);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- stmt = getConnection().createStatement();
- res = stmt.executeQuery(sql);
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, tableName);
+ res = pstmt.executeQuery();
+
exist = res.next();
} catch (SQLException se) {
- throw new IOException(se);
- } finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
+ throw new CatalogException(se);
+ } finally { ㅁ
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
return exist;
}
@Override
- public void addPartition(CatalogProtos.PartitionDescProto proto) throws IOException {
+ public void addPartition(CatalogProtos.PartitionDescProto proto) throws CatalogException {
+ Connection conn = null;
PreparedStatement pstmt = null;
- String sql = "INSERT INTO " + TB_PARTTIONS + " (" + C_TABLE_ID
- + ", partition_method_id, partition_name, ordinal_position, path, cache_nodes) "
- + "VALUES (?, ?, ?, ?, ?, ?, ?) ";
- if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
- }
try {
- pstmt = getConnection().prepareStatement(sql);
+ StringBuilder sql = new StringBuilder();
+ sql.append("INSERT INTO ");
+ sql.append(TB_PARTTIONS);
+ sql.append(" (");
+ sql.append(C_TABLE_ID);
+ sql.append(", partition_method_id, partition_name, ordinal_position, path, cache_nodes) ");
+ sql.append("VALUES (?, ?, ?, ?, ?, ?, ?) ");
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
pstmt.setString(1, proto.getTableId().toLowerCase());
pstmt.setString(2, proto.getPartitionName().toLowerCase());
pstmt.setInt(3, proto.getOrdinalPosition());
@@ -745,38 +900,45 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
pstmt.setString(5, proto.getPath());
pstmt.executeUpdate();
} catch (SQLException se) {
- throw new IOException(se.getMessage(), se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(pstmt);
+ CatalogUtil.closeQuietly(conn, pstmt);
}
}
@Override
- public CatalogProtos.PartitionDescProto getPartition(String partitionName) throws IOException {
+ public CatalogProtos.PartitionDescProto getPartition(String partitionName) throws CatalogException {
+ Connection conn = null;
ResultSet res = null;
- PreparedStatement stmt = null;
+ PreparedStatement pstmt = null;
CatalogProtos.PartitionDescProto proto = null;
- String sql = "SELECT " + C_TABLE_ID
- + ", partition_name, ordinal_position, partition_value, path, cache_nodes FROM "
- + TB_PARTTIONS + "where partition_name = ?";
-
try {
- stmt = getConnection().prepareStatement(sql);
- stmt.setString(1, partitionName);
+ StringBuilder sql = new StringBuilder();
+ sql.append("SELECT ");
+ sql.append(C_TABLE_ID);
+ sql.append(", partition_name, ordinal_position, partition_value, path, cache_nodes FROM ");
+ sql.append(TB_PARTTIONS);
+ sql.append("where partition_name = ?");
+
+
if (LOG.isDebugEnabled()) {
- LOG.debug(stmt.toString());
+ LOG.debug(sql);
}
- res = stmt.executeQuery();
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, partitionName);
+ res = pstmt.executeQuery();
if (!res.next()) {
- throw new IOException("ERROR: there is no index matched to " + partitionName);
+ throw new CatalogException("ERROR: there is no index matched to " + partitionName);
}
proto = resultToPartitionDescProto(res);
} catch (SQLException se) {
- throw new IOException(se.getMessage(), se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
return proto;
@@ -784,32 +946,40 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
@Override
- public CatalogProtos.PartitionsProto getPartitions(String tableName) throws IOException {
+ public CatalogProtos.PartitionsProto getPartitions(String tableName) throws CatalogException {
+ Connection conn = null;
ResultSet res = null;
- PreparedStatement stmt = null;
+ PreparedStatement pstmt = null;
CatalogProtos.PartitionsProto proto = null;
- String partsSql = "SELECT " + C_TABLE_ID
- + ", partition_name, ordinal_position, partition_value, path, cache_nodes FROM "
- + TB_PARTTIONS + "where " + C_TABLE_ID + " = ?";
-
try {
- // PARTITIONS
- stmt = getConnection().prepareStatement(partsSql);
- stmt.setString(1, tableName);
+ StringBuilder sql = new StringBuilder();
+ sql.append("SELECT ");
+ sql.append(C_TABLE_ID);
+ sql.append(", partition_name, ordinal_position, partition_value, path, cache_nodes FROM ");
+ sql.append(TB_PARTTIONS);
+ sql.append("where ");
+ sql.append(C_TABLE_ID);
+ sql.append(" = ?");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(stmt.toString());
+ LOG.debug(sql.toString());
}
- res = stmt.executeQuery();
+
+ // PARTITIONS
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, tableName);
+ res = pstmt.executeQuery();
CatalogProtos.PartitionsProto.Builder builder = CatalogProtos.PartitionsProto.newBuilder();
while(res.next()) {
builder.addPartition(resultToPartitionDescProto(res));
}
proto = builder.build();
} catch (SQLException se) {
- throw new IOException(se.getMessage(), se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
return proto;
@@ -817,132 +987,149 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
@Override
- public void delPartition(String partitionName) throws IOException {
- String sql =
- "DELETE FROM " + TB_PARTTIONS
- + " WHERE partition_name ='" + partitionName + "'";
- if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
- }
- Statement stmt = null;
+ public void delPartition(String partitionName) throws CatalogException {
+ Connection conn = null;
+ PreparedStatement pstmt = null;
try {
- stmt = getConnection().createStatement();
+ StringBuilder sql = new StringBuilder();
+ sql.append(" DELETE FROM ").append(TB_PARTTIONS);
+ sql.append(" WHERE partition_name = ? ");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- stmt.executeUpdate(sql);
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, partitionName);
+ pstmt.executeUpdate();
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(stmt);
+ CatalogUtil.closeQuietly(conn, pstmt);
}
}
@Override
- public void delPartitions(String tableName) throws IOException {
- String sql =
- "DELETE FROM " + TB_PARTTIONS
- + " WHERE " + C_TABLE_ID + " ='" + tableName + "'";
- if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
- }
- Statement stmt = null;
+ public void delPartitions(String tableName) throws CatalogException {
+ Connection conn = null;
+ PreparedStatement pstmt = null;
try {
- stmt = getConnection().createStatement();
+ StringBuilder sql = new StringBuilder();
+ sql.append(" DELETE FROM ").append(TB_PARTTIONS);
+ sql.append(" WHERE ").append(C_TABLE_ID).append(" = ? ");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- stmt.executeUpdate(sql);
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, tableName);
+ pstmt.executeUpdate();
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(stmt);
+ CatalogUtil.closeQuietly(conn, pstmt);
}
}
@Override
- public void addIndex(final IndexDescProto proto) throws IOException {
- String sql =
- "INSERT INTO indexes (index_name, " + C_TABLE_ID + ", column_name, "
- + "data_type, index_type, is_unique, is_clustered, is_ascending) VALUES "
- + "(?,?,?,?,?,?,?,?)";
-
- PreparedStatement stmt = null;
+ public void addIndex(final IndexDescProto proto) throws CatalogException {
+ Connection conn = null;
+ PreparedStatement pstmt = null;
try {
- stmt = getConnection().prepareStatement(sql);
- stmt.setString(1, proto.getName());
- stmt.setString(2, proto.getTableId());
- stmt.setString(3, proto.getColumn().getColumnName());
- stmt.setString(4, proto.getColumn().getDataType().getType().name());
- stmt.setString(5, proto.getIndexMethod().toString());
- stmt.setBoolean(6, proto.hasIsUnique() && proto.getIsUnique());
- stmt.setBoolean(7, proto.hasIsClustered() && proto.getIsClustered());
- stmt.setBoolean(8, proto.hasIsAscending() && proto.getIsAscending());
- stmt.executeUpdate();
+ StringBuilder sql = new StringBuilder();
+ sql.append("INSERT INTO indexes (index_name, ");
+ sql.append(C_TABLE_ID);
+ sql.append(", column_name, ");
+ sql.append( "data_type, index_type, is_unique, is_clustered, is_ascending) VALUES ");
+ sql.append("(?,?,?,?,?,?,?,?)");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(stmt.toString());
+ LOG.debug(sql.toString());
}
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+
+ pstmt.setString(1, proto.getName());
+ pstmt.setString(2, proto.getTableId());
+ pstmt.setString(3, proto.getColumn().getColumnName());
+ pstmt.setString(4, proto.getColumn().getDataType().getType().name());
+ pstmt.setString(5, proto.getIndexMethod().toString());
+ pstmt.setBoolean(6, proto.hasIsUnique() && proto.getIsUnique());
+ pstmt.setBoolean(7, proto.hasIsClustered() && proto.getIsClustered());
+ pstmt.setBoolean(8, proto.hasIsAscending() && proto.getIsAscending());
+ pstmt.executeUpdate();
+
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(stmt);
+ CatalogUtil.closeQuietly(conn, pstmt);
}
}
@Override
- public void delIndex(final String indexName) throws IOException {
- String sql =
- "DELETE FROM " + TB_INDEXES
- + " WHERE index_name='" + indexName + "'";
- if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
- }
-
- Statement stmt = null;
+ public void delIndex(final String indexName) throws CatalogException {
+ Connection conn = null;
+ PreparedStatement pstmt = null;
try {
- stmt = getConnection().createStatement();
+ StringBuilder sql = new StringBuilder();
+ sql.append(" DELETE FROM ").append(TB_INDEXES);
+ sql.append(" WHERE index_name ").append(" = ? ");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- stmt.executeUpdate(sql);
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, indexName);
+ pstmt.executeUpdate();
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(stmt);
+ CatalogUtil.closeQuietly(conn, pstmt);
}
}
@Override
- public IndexDescProto getIndex(final String indexName)
- throws IOException {
+ public IndexDescProto getIndex(final String indexName) throws CatalogException {
+ Connection conn = null;
ResultSet res = null;
- PreparedStatement stmt = null;
-
+ PreparedStatement pstmt = null;
IndexDescProto proto = null;
try {
- String sql =
- "SELECT index_name, " + C_TABLE_ID + ", column_name, data_type, "
- + "index_type, is_unique, is_clustered, is_ascending FROM indexes "
- + "where index_name = ?";
- stmt = getConnection().prepareStatement(sql);
- stmt.setString(1, indexName);
+ StringBuilder sql = new StringBuilder();
+ sql.append( "SELECT index_name, ");
+ sql.append(C_TABLE_ID);
+ sql.append(", column_name, data_type, ");
+ sql.append("index_type, is_unique, is_clustered, is_ascending FROM indexes ");
+ sql.append("where index_name = ?");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(stmt.toString());
+ LOG.debug(sql.toString());
}
- res = stmt.executeQuery();
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, indexName);
+ res = pstmt.executeQuery();
if (!res.next()) {
- throw new IOException("ERROR: there is no index matched to " + indexName);
+ throw new CatalogException("ERROR: there is no index matched to " + indexName);
}
proto = resultToIndexDescProto(res);
} catch (SQLException se) {
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
return proto;
@@ -950,62 +1137,75 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
@Override
public IndexDescProto getIndex(final String tableName,
- final String columnName) throws IOException {
+ final String columnName) throws CatalogException {
+ Connection conn = null;
ResultSet res = null;
- PreparedStatement stmt = null;
-
+ PreparedStatement pstmt = null;
IndexDescProto proto = null;
try {
- String sql =
- "SELECT index_name, " + C_TABLE_ID + ", column_name, data_type, "
- + "index_type, is_unique, is_clustered, is_ascending FROM indexes "
- + "where " + C_TABLE_ID + " = ? AND column_name = ?";
- stmt = getConnection().prepareStatement(sql);
- stmt.setString(1, tableName);
- stmt.setString(2, columnName);
+ StringBuilder sql = new StringBuilder();
+ sql.append("SELECT index_name, ");
+ sql.append(C_TABLE_ID);
+ sql.append(", column_name, data_type, ");
+ sql.append("index_type, is_unique, is_clustered, is_ascending FROM indexes ");
+ sql.append("where ");
+ sql.append(C_TABLE_ID);
+ sql.append(" = ? AND column_name = ?");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- res = stmt.executeQuery();
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, tableName);
+ pstmt.setString(2, columnName);
+
+ res = pstmt.executeQuery();
if (!res.next()) {
- throw new IOException("ERROR: there is no index matched to "
+ throw new CatalogException("ERROR: there is no index matched to "
+ tableName + "." + columnName);
}
proto = resultToIndexDescProto(res);
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
return proto;
}
@Override
- public boolean existIndex(final String indexName) throws IOException {
- String sql = "SELECT index_name from " + TB_INDEXES
- + " WHERE index_name = ?";
- if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
- }
-
- PreparedStatement stmt = null;
+ public boolean existIndex(final String indexName) throws CatalogException {
+ Connection conn = null;
+ PreparedStatement pstmt = null;
ResultSet res = null;
boolean exist = false;
try {
- stmt = getConnection().prepareStatement(sql);
- stmt.setString(1, indexName);
+ StringBuilder sql = new StringBuilder();
+ sql.append("SELECT index_name from ");
+ sql.append(TB_INDEXES);
+ sql.append(" WHERE index_name = ?");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- res = stmt.executeQuery();
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, indexName);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
+ res = pstmt.executeQuery();
exist = res.next();
} catch (SQLException se) {
} finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
return exist;
@@ -1013,30 +1213,37 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
@Override
public boolean existIndex(String tableName, String columnName)
- throws IOException {
- String sql = "SELECT index_name from " + TB_INDEXES
- + " WHERE " + C_TABLE_ID + " = ? AND COLUMN_NAME = ?";
- if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
- }
-
- PreparedStatement stmt = null;
+ throws CatalogException {
+ Connection conn = null;
+ PreparedStatement pstmt = null;
boolean exist = false;
ResultSet res = null;
try {
- stmt = getConnection().prepareStatement(sql);
- stmt.setString(1, tableName);
- stmt.setString(2, columnName);
+ StringBuilder sql = new StringBuilder();
+ sql.append("SELECT index_name from ");
+ sql.append(TB_INDEXES);
+ sql.append(" WHERE ");
+ sql.append(C_TABLE_ID);
+ sql.append(" = ? AND COLUMN_NAME = ?");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- res = stmt.executeQuery();
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, tableName);
+ pstmt.setString(2, columnName);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sql.toString());
+ }
+ res = pstmt.executeQuery();
exist = res.next();
} catch (SQLException se) {
-
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
return exist;
@@ -1044,28 +1251,39 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
@Override
public IndexDescProto[] getIndexes(final String tableName)
- throws IOException {
+ throws CatalogException {
+ Connection conn = null;
ResultSet res = null;
- PreparedStatement stmt = null;
-
- List<IndexDescProto> protos = new ArrayList<IndexDescProto>();
+ PreparedStatement pstmt = null;
+ List<IndexDescProto> protos = null;
try {
- String sql = "SELECT index_name, " + C_TABLE_ID + ", column_name, data_type, "
- + "index_type, is_unique, is_clustered, is_ascending FROM indexes "
- + "where " + C_TABLE_ID + "= ?";
- stmt = getConnection().prepareStatement(sql);
- stmt.setString(1, tableName);
+ protos = new ArrayList<IndexDescProto>();
+
+ StringBuilder sql = new StringBuilder();
+ sql.append("SELECT index_name, ");
+ sql.append(C_TABLE_ID);
+ sql.append(", column_name, data_type, ");
+ sql.append("index_type, is_unique, is_clustered, is_ascending FROM indexes ");
+ sql.append("where ");
+ sql.append(C_TABLE_ID);
+ sql.append("= ?");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(sql);
+ LOG.debug(sql.toString());
}
- res = stmt.executeQuery();
+
+ conn = getConnection();
+ pstmt = conn.prepareStatement(sql.toString());
+ pstmt.setString(1, tableName);
+ res = pstmt.executeQuery();
while (res.next()) {
protos.add(resultToIndexDescProto(res));
}
} catch (SQLException se) {
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(res, stmt);
+ CatalogUtil.closeQuietly(conn, pstmt, res);
}
return protos.toArray(new IndexDescProto[protos.size()]);
@@ -1173,30 +1391,28 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
@Override
public void close() {
- CatalogUtil.closeSQLWrapper(conn);
+ CatalogUtil.closeQuietly(conn);
LOG.info("Shutdown database (" + catalogUri + ")");
}
@Override
- public final void addFunction(final FunctionDesc func) throws IOException {
+ public final void addFunction(final FunctionDesc func) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public final void deleteFunction(final FunctionDesc func) throws IOException {
+ public final void deleteFunction(final FunctionDesc func) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public final void existFunction(final FunctionDesc func) throws IOException {
+ public final void existFunction(final FunctionDesc func) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public final List<String> getAllFunctionNames() throws IOException {
+ public final List<String> getAllFunctionNames() throws CatalogException {
// TODO - not implemented yet
return null;
}
-
-
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java
index 8a824bd..db15c02 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java
@@ -23,73 +23,72 @@ import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.proto.CatalogProtos.IndexDescProto;
import java.io.Closeable;
-import java.io.IOException;
+import org.apache.tajo.catalog.exception.CatalogException;
import java.util.List;
public interface CatalogStore extends Closeable {
/*************************** TABLE ******************************/
- void addTable(CatalogProtos.TableDescProto desc) throws IOException;
+ void addTable(CatalogProtos.TableDescProto desc) throws CatalogException;
- boolean existTable(String name) throws IOException;
+ boolean existTable(String name) throws CatalogException;
- void deleteTable(String name) throws IOException;
+ void deleteTable(String name) throws CatalogException;
- CatalogProtos.TableDescProto getTable(String name) throws IOException;
+ CatalogProtos.TableDescProto getTable(String name) throws CatalogException;
- List<String> getAllTableNames() throws IOException;
+ List<String> getAllTableNames() throws CatalogException;
/************************ PARTITION METHOD **************************/
- void addPartitionMethod(CatalogProtos.PartitionMethodProto partitionMethodProto) throws IOException;
+ void addPartitionMethod(CatalogProtos.PartitionMethodProto partitionMethodProto) throws CatalogException;
- CatalogProtos.PartitionMethodProto getPartitionMethod(String tableName) throws IOException;
+ CatalogProtos.PartitionMethodProto getPartitionMethod(String tableName) throws CatalogException;
- boolean existPartitionMethod(String tableName) throws IOException;
+ boolean existPartitionMethod(String tableName) throws CatalogException;
- void delPartitionMethod(String tableName) throws IOException;
+ void delPartitionMethod(String tableName) throws CatalogException;
/************************** PARTITIONS *****************************/
- void addPartitions(CatalogProtos.PartitionsProto partitionsProto) throws IOException;
+ void addPartitions(CatalogProtos.PartitionsProto partitionsProto) throws CatalogException;
- void addPartition(CatalogProtos.PartitionDescProto partitionDescProto) throws IOException;
+ void addPartition(CatalogProtos.PartitionDescProto partitionDescProto) throws CatalogException;
/**
* Get all partitions of a table
* @param tableName the table name
* @return
- * @throws IOException
+ * @throws CatalogException
*/
- CatalogProtos.PartitionsProto getPartitions(String tableName) throws IOException;
+ CatalogProtos.PartitionsProto getPartitions(String tableName) throws CatalogException;
- CatalogProtos.PartitionDescProto getPartition(String partitionName) throws IOException;
+ CatalogProtos.PartitionDescProto getPartition(String partitionName) throws CatalogException;
- void delPartition(String partitionName) throws IOException;
+ void delPartition(String partitionName) throws CatalogException;
- void delPartitions(String tableName) throws IOException;
+ void delPartitions(String tableName) throws CatalogException;
/**************************** INDEX *******************************/
- void addIndex(IndexDescProto proto) throws IOException;
+ void addIndex(IndexDescProto proto) throws CatalogException;
- void delIndex(String indexName) throws IOException;
+ void delIndex(String indexName) throws CatalogException;
- IndexDescProto getIndex(String indexName) throws IOException;
+ IndexDescProto getIndex(String indexName) throws CatalogException;
- IndexDescProto getIndex(String tableName, String columnName)
- throws IOException;
+ IndexDescProto getIndex(String tableName, String columnName) throws CatalogException;
- boolean existIndex(String indexName) throws IOException;
+ boolean existIndex(String indexName) throws CatalogException;
- boolean existIndex(String tableName, String columnName) throws IOException;
+ boolean existIndex(String tableName, String columnName) throws CatalogException;
/************************** FUNCTION *****************************/
- IndexDescProto [] getIndexes(String tableName) throws IOException;
+ IndexDescProto [] getIndexes(String tableName) throws CatalogException;
- void addFunction(FunctionDesc func) throws IOException;
+ void addFunction(FunctionDesc func) throws CatalogException;
- void deleteFunction(FunctionDesc func) throws IOException;
+ void deleteFunction(FunctionDesc func) throws CatalogException;
- void existFunction(FunctionDesc func) throws IOException;
+ void existFunction(FunctionDesc func) throws CatalogException;
- List<String> getAllFunctionNames() throws IOException;
+ List<String> getAllFunctionNames() throws CatalogException;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/DerbyStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/DerbyStore.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/DerbyStore.java
index 6f1b612..b4bd605 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/DerbyStore.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/DerbyStore.java
@@ -17,20 +17,17 @@
*/
/**
- *
+ *
*/
package org.apache.tajo.catalog.store;
import org.apache.hadoop.conf.Configuration;
import org.apache.tajo.catalog.CatalogUtil;
-import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.catalog.exception.CatalogException;
import org.apache.tajo.exception.InternalException;
-import java.io.IOException;
import java.sql.*;
-
import java.util.Map;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
public class DerbyStore extends AbstractDBStore {
private static final String CATALOG_DRIVER="org.apache.derby.jdbc.EmbeddedDriver";
@@ -38,7 +35,7 @@ public class DerbyStore extends AbstractDBStore {
protected String getCatalogDriverName(){
return CATALOG_DRIVER;
}
-
+
public DerbyStore(final Configuration conf)
throws InternalException {
super(conf);
@@ -49,229 +46,294 @@ public class DerbyStore extends AbstractDBStore {
}
// TODO - DDL and index statements should be renamed
- protected void createBaseTable() throws IOException {
+ protected void createBaseTable() throws CatalogException {
+ Connection conn = null;
Statement stmt = null;
+
try {
- // META
- stmt = getConnection().createStatement();
+ conn = getConnection();
+ stmt = conn.createStatement();
+
+ StringBuilder sql = new StringBuilder();
+ //META
if (!baseTableMaps.get(TB_META)) {
- String meta_ddl = "CREATE TABLE " + TB_META + " (version int NOT NULL)";
+ sql.append("CREATE TABLE ");
+ sql.append(TB_META);
+ sql.append(" (version int NOT NULL)");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(meta_ddl);
+ LOG.debug(sql.toString());
}
- stmt.executeUpdate(meta_ddl);
+ stmt.executeUpdate(sql.toString());
LOG.info("Table '" + TB_META + " is created.");
}
// TABLES
if (!baseTableMaps.get(TB_TABLES)) {
- String tables_ddl = "CREATE TABLE "
- + TB_TABLES + " ("
- + "TID int NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1), "
- + C_TABLE_ID + " VARCHAR(255) NOT NULL CONSTRAINT TABLE_ID_UNIQ UNIQUE, "
- + "path VARCHAR(1024), "
- + "store_type CHAR(16), "
- + "CONSTRAINT TABLES_PK PRIMARY KEY (TID)" +
- ")";
+ sql.delete(0, sql.length());
+ sql.append("CREATE TABLE ");
+ sql.append(TB_TABLES);
+ sql.append(" (");
+ sql.append("TID int NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1),");
+ sql.append(C_TABLE_ID);
+ sql.append(" VARCHAR(255) NOT NULL CONSTRAINT TABLE_ID_UNIQ UNIQUE, ");
+ sql.append("path VARCHAR(1024), ");
+ sql.append("store_type CHAR(16), ");
+ sql.append("CONSTRAINT TABLES_PK PRIMARY KEY (TID)");
+ sql.append( ")");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(tables_ddl);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(tables_ddl);
+ stmt.addBatch(sql.toString());
+
+ sql.delete(0, sql.length());
+ sql.append("CREATE UNIQUE INDEX idx_tables_tid on ");
+ sql.append(TB_TABLES);
+ sql.append(" (TID)");
- String idx_tables_tid =
- "CREATE UNIQUE INDEX idx_tables_tid on " + TB_TABLES + " (TID)";
if (LOG.isDebugEnabled()) {
- LOG.debug(idx_tables_tid);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(idx_tables_tid);
+ stmt.addBatch(sql.toString());
+
+ sql.delete(0, sql.length());
+ sql.append("CREATE UNIQUE INDEX idx_tables_name on ");
+ sql.append(TB_TABLES);
+ sql.append("(");
+ sql.append(C_TABLE_ID);
+ sql.append(")");
- String idx_tables_name = "CREATE UNIQUE INDEX idx_tables_name on "
- + TB_TABLES + "(" + C_TABLE_ID + ")";
if (LOG.isDebugEnabled()) {
- LOG.debug(idx_tables_name);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(idx_tables_name);
+ stmt.addBatch(sql.toString());
stmt.executeBatch();
LOG.info("Table '" + TB_TABLES + "' is created.");
+
}
// COLUMNS
if (!baseTableMaps.get(TB_COLUMNS)) {
- String columns_ddl =
- "CREATE TABLE " + TB_COLUMNS + " ("
- + "TID INT NOT NULL REFERENCES " + TB_TABLES + " (TID) ON DELETE CASCADE, "
- + C_TABLE_ID + " VARCHAR(255) NOT NULL REFERENCES " + TB_TABLES + "("
- + C_TABLE_ID + ") ON DELETE CASCADE, "
- + "column_id INT NOT NULL,"
- + "column_name VARCHAR(255) NOT NULL, " + "data_type CHAR(16), " + "type_length INTEGER, "
- + "CONSTRAINT C_COLUMN_ID UNIQUE (" + C_TABLE_ID + ", column_name))";
+ sql.delete(0, sql.length());
+ sql.append("CREATE TABLE ");
+ sql.append(TB_COLUMNS);
+ sql.append(" (");
+ sql.append("TID INT NOT NULL REFERENCES ");
+ sql.append(TB_TABLES);
+ sql.append(" (TID) ON DELETE CASCADE, ");
+ sql.append(C_TABLE_ID);
+ sql.append( " VARCHAR(255) NOT NULL REFERENCES ");
+ sql.append(TB_TABLES);
+ sql.append("(");
+ sql.append(C_TABLE_ID);
+ sql.append(") ON DELETE CASCADE, ");
+ sql.append("column_id INT NOT NULL,");
+ sql.append("column_name VARCHAR(255) NOT NULL, ");
+ sql.append("data_type CHAR(16), type_length INTEGER, ");
+ sql.append("CONSTRAINT C_COLUMN_ID UNIQUE (");
+ sql.append(C_TABLE_ID);
+ sql.append(", column_name))");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(columns_ddl);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(columns_ddl);
+ stmt.addBatch(sql.toString());
+
+ sql.delete(0, sql.length());
+ sql.append( "CREATE UNIQUE INDEX idx_fk_columns_table_name on ");
+ sql.append(TB_COLUMNS);
+ sql.append("(");
+ sql.append(C_TABLE_ID);
+ sql.append(", column_name)");
- String idx_fk_columns_table_name =
- "CREATE UNIQUE INDEX idx_fk_columns_table_name on "
- + TB_COLUMNS + "(" + C_TABLE_ID + ", column_name)";
if (LOG.isDebugEnabled()) {
- LOG.debug(idx_fk_columns_table_name);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(idx_fk_columns_table_name);
+ stmt.addBatch(sql.toString());
stmt.executeBatch();
LOG.info("Table '" + TB_COLUMNS + " is created.");
}
// OPTIONS
if (!baseTableMaps.get(TB_OPTIONS)) {
- String options_ddl =
- "CREATE TABLE " + TB_OPTIONS +" ("
- + C_TABLE_ID + " VARCHAR(255) NOT NULL REFERENCES TABLES (" + C_TABLE_ID +") "
- + "ON DELETE CASCADE, "
- + "key_ VARCHAR(255) NOT NULL, value_ VARCHAR(255) NOT NULL)";
+ sql.delete(0, sql.length());
+ sql.append( "CREATE TABLE ");
+ sql.append(TB_OPTIONS);
+ sql.append(" (").append(C_TABLE_ID);
+ sql.append(" VARCHAR(255) NOT NULL REFERENCES TABLES (");
+ sql.append(C_TABLE_ID).append(") ON DELETE CASCADE, ");
+ sql.append("key_ VARCHAR(255) NOT NULL, value_ VARCHAR(255) NOT NULL)");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(options_ddl);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(options_ddl);
+ stmt.addBatch(sql.toString());
+
+ sql.delete(0, sql.length());
+ sql.append("CREATE INDEX idx_options_key on ");
+ sql.append(TB_OPTIONS).append( " (").append(C_TABLE_ID).append(")");
- String idx_options_key =
- "CREATE INDEX idx_options_key on " + TB_OPTIONS + " (" + C_TABLE_ID + ")";
if (LOG.isDebugEnabled()) {
- LOG.debug(idx_options_key);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(idx_options_key);
- String idx_options_table_name =
- "CREATE INDEX idx_options_table_name on " + TB_OPTIONS
- + "(" + C_TABLE_ID + ")";
+ stmt.addBatch(sql.toString());
+
+ sql.delete(0, sql.length());
+ sql.append("CREATE INDEX idx_options_table_name on ").append(TB_OPTIONS);
+ sql.append("(" ).append(C_TABLE_ID).append(")");
if (LOG.isDebugEnabled()) {
- LOG.debug(idx_options_table_name);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(idx_options_table_name);
+ stmt.addBatch(sql.toString());
stmt.executeBatch();
LOG.info("Table '" + TB_OPTIONS + " is created.");
}
// INDEXES
if (!baseTableMaps.get(TB_INDEXES)) {
- String indexes_ddl = "CREATE TABLE " + TB_INDEXES +"("
- + "index_name VARCHAR(255) NOT NULL PRIMARY KEY, "
- + C_TABLE_ID + " VARCHAR(255) NOT NULL REFERENCES TABLES (" + C_TABLE_ID + ") "
- + "ON DELETE CASCADE, "
- + "column_name VARCHAR(255) NOT NULL, "
- + "data_type VARCHAR(255) NOT NULL, "
- + "index_type CHAR(32) NOT NULL, "
- + "is_unique BOOLEAN NOT NULL, "
- + "is_clustered BOOLEAN NOT NULL, "
- + "is_ascending BOOLEAN NOT NULL)";
+ sql.delete(0, sql.length());
+ sql.append("CREATE TABLE ").append(TB_INDEXES).append("(");
+ sql.append( "index_name VARCHAR(255) NOT NULL PRIMARY KEY, ");
+ sql.append(C_TABLE_ID).append(" VARCHAR(255) NOT NULL REFERENCES TABLES (");
+ sql.append(C_TABLE_ID).append(") ");
+ sql.append("ON DELETE CASCADE, ");
+ sql.append("column_name VARCHAR(255) NOT NULL, ");
+ sql.append("data_type VARCHAR(255) NOT NULL, ");
+ sql.append("index_type CHAR(32) NOT NULL, ");
+ sql.append("is_unique BOOLEAN NOT NULL, ");
+ sql.append("is_clustered BOOLEAN NOT NULL, ");
+ sql.append("is_ascending BOOLEAN NOT NULL)");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(indexes_ddl);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(indexes_ddl);
+ stmt.addBatch(sql.toString());
+
+ sql.delete(0, sql.length());
+ sql.append("CREATE UNIQUE INDEX idx_indexes_key ON ");
+ sql.append(TB_INDEXES).append(" (index_name)");
- String idx_indexes_key = "CREATE UNIQUE INDEX idx_indexes_key ON "
- + TB_INDEXES + " (index_name)";
if (LOG.isDebugEnabled()) {
- LOG.debug(idx_indexes_key);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(idx_indexes_key);
+ stmt.addBatch(sql.toString());
+
+ sql.delete(0, sql.length());
+ sql.append("CREATE INDEX idx_indexes_columns ON ");
+ sql.append(TB_INDEXES).append(" (").append(C_TABLE_ID).append(", column_name)");
- String idx_indexes_columns = "CREATE INDEX idx_indexes_columns ON "
- + TB_INDEXES + " (" + C_TABLE_ID + ", column_name)";
if (LOG.isDebugEnabled()) {
- LOG.debug(idx_indexes_columns);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(idx_indexes_columns);
+ stmt.addBatch(sql.toString());
stmt.executeBatch();
LOG.info("Table '" + TB_INDEXES + "' is created.");
}
if (!baseTableMaps.get(TB_STATISTICS)) {
- String stats_ddl = "CREATE TABLE " + TB_STATISTICS + "("
- + C_TABLE_ID + " VARCHAR(255) NOT NULL REFERENCES TABLES (" + C_TABLE_ID + ") "
- + "ON DELETE CASCADE, "
- + "num_rows BIGINT, "
- + "num_bytes BIGINT)";
+ sql.delete(0, sql.length());
+ sql.append("CREATE TABLE ").append(TB_STATISTICS).append( "(");
+ sql.append(C_TABLE_ID).append(" VARCHAR(255) NOT NULL REFERENCES TABLES (");
+ sql.append(C_TABLE_ID).append(") ");
+ sql.append("ON DELETE CASCADE, ");
+ sql.append("num_rows BIGINT, ");
+ sql.append("num_bytes BIGINT)");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(stats_ddl);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(stats_ddl);
+ stmt.addBatch(sql.toString());
+
+ sql.delete(0, sql.length());
+ sql.append("CREATE INDEX idx_stats_table_name ON ");
+ sql.append(TB_STATISTICS).append(" (").append(C_TABLE_ID).append(")");
- String idx_stats_fk_table_name = "CREATE INDEX idx_stats_table_name ON "
- + TB_STATISTICS + " (" + C_TABLE_ID + ")";
if (LOG.isDebugEnabled()) {
- LOG.debug(idx_stats_fk_table_name);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(idx_stats_fk_table_name);
+ stmt.addBatch(sql.toString());
stmt.executeBatch();
LOG.info("Table '" + TB_STATISTICS + "' is created.");
}
// PARTITION_METHODS
if (!baseTableMaps.get(TB_PARTITION_METHODS)) {
- String partition_method_ddl = "CREATE TABLE " + TB_PARTITION_METHODS + " ("
- + C_TABLE_ID + " VARCHAR(255) NOT NULL REFERENCES TABLES (" + C_TABLE_ID + ") "
- + "ON DELETE CASCADE, "
- + "partition_type VARCHAR(10) NOT NULL,"
- + "expression VARCHAR(1024) NOT NULL,"
- + "expression_schema VARCHAR(1024) FOR BIT DATA NOT NULL)";
+ sql.delete(0, sql.length());
+ sql.append("CREATE TABLE ").append(TB_PARTITION_METHODS).append(" (");
+ sql.append(C_TABLE_ID).append(" VARCHAR(255) NOT NULL REFERENCES TABLES (");
+ sql.append(C_TABLE_ID).append(") ");
+ sql.append("ON DELETE CASCADE, ");
+ sql.append("partition_type VARCHAR(10) NOT NULL,");
+ sql.append("expression VARCHAR(1024) NOT NULL,");
+ sql.append("expression_schema VARCHAR(1024) FOR BIT DATA NOT NULL)");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(partition_method_ddl);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(partition_method_ddl);
+ stmt.addBatch(sql.toString());
+
+ sql.delete(0, sql.length());
+ sql.append("CREATE INDEX idx_partition_methods_table_name ON ");
+ sql.append(TB_PARTITION_METHODS).append(" (").append(C_TABLE_ID).append(")");
- String idx_partition_methods_table_name = "CREATE INDEX idx_partition_methods_table_name ON "
- + TB_PARTITION_METHODS + " (" + C_TABLE_ID + ")";
if (LOG.isDebugEnabled()) {
- LOG.debug(idx_partition_methods_table_name);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(idx_partition_methods_table_name);
+ stmt.addBatch(sql.toString());
stmt.executeBatch();
LOG.info("Table '" + TB_PARTITION_METHODS + "' is created.");
}
// PARTITIONS
if (!baseTableMaps.get(TB_PARTTIONS)) {
- String partition_ddl = "CREATE TABLE " + TB_PARTTIONS + " ("
- + "PID INT NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1), "
- + C_TABLE_ID + " VARCHAR(255) NOT NULL REFERENCES TABLES (" + C_TABLE_ID + ") "
- + "ON DELETE CASCADE, "
- + "partition_name VARCHAR(255), "
- + "ordinal_position INT NOT NULL,"
- + "partition_value VARCHAR(1024),"
- + "path VARCHAR(1024),"
- + "cache_nodes VARCHAR(255), "
- + " CONSTRAINT PARTITION_PK PRIMARY KEY (PID)"
- + " )";
+ sql.delete(0, sql.length());
+ sql.append("CREATE TABLE ").append(TB_PARTTIONS).append("(");
+ sql.append("PID INT NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1),");
+ sql.append(C_TABLE_ID).append(" VARCHAR(255) NOT NULL REFERENCES TABLES (");
+ sql.append(C_TABLE_ID).append(")");
+ sql.append("ON DELETE CASCADE, ");
+ sql.append("partition_name VARCHAR(255), ");
+ sql.append("ordinal_position INT NOT NULL,");
+ sql.append("partition_value VARCHAR(1024),");
+ sql.append("path VARCHAR(1024),");
+ sql.append("cache_nodes VARCHAR(255), ");
+ sql.append(" CONSTRAINT PARTITION_PK PRIMARY KEY (PID))");
+
if (LOG.isDebugEnabled()) {
- LOG.debug(partition_ddl);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(partition_ddl);
+ stmt.addBatch(sql.toString());
+
+ sql.delete(0, sql.length());
+ sql.append("CREATE INDEX idx_partitions_table_name ON ");
+ sql.append(TB_PARTTIONS).append(" (").append(C_TABLE_ID).append(")");
- String idx_partitions_table_name = "CREATE INDEX idx_partitions_table_name ON "
- + TB_PARTTIONS + " (" + C_TABLE_ID + ")";
if (LOG.isDebugEnabled()) {
- LOG.debug(idx_partitions_table_name);
+ LOG.debug(sql.toString());
}
- stmt.addBatch(idx_partitions_table_name);
+ stmt.addBatch(sql.toString());
stmt.executeBatch();
LOG.info("Table '" + TB_PARTTIONS + "' is created.");
}
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(stmt);
+ CatalogUtil.closeQuietly(conn, stmt);
}
}
@Override
- protected boolean isInitialized() throws IOException {
- ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
-
+ protected boolean isInitialized() throws CatalogException {
+ Connection conn = null;
ResultSet res = null;
int foundCount = 0;
+
try {
- res = getConnection().getMetaData().getTables(null, null, null,
+ conn = getConnection();
+ res = conn.getMetaData().getTables(null, null, null,
new String [] {"TABLE"});
baseTableMaps.put(TB_META, false);
@@ -287,9 +349,9 @@ public class DerbyStore extends AbstractDBStore {
baseTableMaps.put(res.getString("TABLE_NAME"), true);
}
} catch (SQLException se){
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(res);
+ CatalogUtil.closeQuietly(conn, res);
}
for(Map.Entry<String, Boolean> entry : baseTableMaps.entrySet()) {
@@ -300,42 +362,48 @@ public class DerbyStore extends AbstractDBStore {
return true;
}
-
- final boolean checkInternalTable(final String tableName) throws IOException {
+
+ final boolean checkInternalTable(final String tableName) throws CatalogException {
+ Connection conn = null;
ResultSet res = null;
+ boolean found = false;
+
try {
- boolean found = false;
- res = getConnection().getMetaData().getTables(null, null, null,
- new String [] {"TABLE"});
+ conn = getConnection();
+ res = conn.getMetaData().getTables(null, null, null,
+ new String [] {"TABLE"});
while(res.next() && !found) {
if (tableName.equals(res.getString("TABLE_NAME")))
found = true;
}
-
- return found;
+
} catch (SQLException se) {
- throw new IOException(se);
+ throw new CatalogException(se);
} finally {
- CatalogUtil.closeSQLWrapper(res);
+ CatalogUtil.closeQuietly(conn, res);
}
+ return found;
}
-
+
@Override
public final void close() {
+ Connection conn = null;
+ // shutdown embedded database.
try {
- DriverManager.getConnection("jdbc:derby:;shutdown=true");
- } catch (SQLException e) {
- // TODO - to be fixed
- //LOG.error(e.getMessage(), e);
+ // the shutdown=true attribute shuts down Derby.
+ conn = DriverManager.getConnection("jdbc:derby:;shutdown=true");
+ } catch (SQLException se) {
+ if ( (se.getErrorCode() == 50000)
+ && (se.getSQLState().equals("XJ015"))) {
+ // tajo got the expected exception
+ LOG.info("Derby shutdown complete normally.");
+ } else {
+ LOG.info("Derby shutdown complete abnormally. - message:" + se.getMessage());
+ }
+ } finally {
+ CatalogUtil.closeQuietly(conn);
}
-
LOG.info("Shutdown database (" + catalogUri + ")");
}
-
-
- public static void main(final String[] args) throws IOException {
- @SuppressWarnings("unused")
- DerbyStore store = new DerbyStore(new TajoConf());
- }
}
[3/3] git commit: TAJO-530: Fix warnings in tajo-catalog. (jaehwa)
Posted by bl...@apache.org.
TAJO-530: Fix warnings in tajo-catalog. (jaehwa)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tajo/commit/413a4b3f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tajo/tree/413a4b3f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tajo/diff/413a4b3f
Branch: refs/heads/master
Commit: 413a4b3fc9bdfa49f4d58d9414f58c30a169c5ab
Parents: c586227
Author: blrunner <jh...@gruter.com>
Authored: Thu Feb 20 00:02:46 2014 +0900
Committer: blrunner <jh...@gruter.com>
Committed: Thu Feb 20 00:02:46 2014 +0900
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../org/apache/tajo/catalog/CatalogUtil.java | 70 +-
.../org/apache/tajo/catalog/FunctionDesc.java | 16 +-
.../java/org/apache/tajo/catalog/Options.java | 11 +-
.../java/org/apache/tajo/catalog/Schema.java | 20 +-
.../java/org/apache/tajo/catalog/SortSpec.java | 6 +
.../java/org/apache/tajo/catalog/TableDesc.java | 5 +
.../tajo/catalog/json/TableMetaAdapter.java | 8 +-
.../tajo/catalog/partition/PartitionDesc.java | 18 +
.../catalog/partition/PartitionMethodDesc.java | 6 +
.../apache/tajo/catalog/statistics/StatSet.java | 21 +-
.../tajo/catalog/store/HCatalogStore.java | 240 ++--
.../catalog/store/HCatalogStoreClientPool.java | 26 +-
.../apache/tajo/catalog/store/HCatalogUtil.java | 39 +-
.../org/apache/tajo/catalog/CatalogServer.java | 97 +-
.../tajo/catalog/store/AbstractDBStore.java | 1140 +++++++++++-------
.../apache/tajo/catalog/store/CatalogStore.java | 57 +-
.../apache/tajo/catalog/store/DerbyStore.java | 368 +++---
.../org/apache/tajo/catalog/store/MemStore.java | 71 +-
.../apache/tajo/catalog/store/MySQLStore.java | 208 ++--
.../tajo/engine/planner/global/DataChannel.java | 3 +-
.../planner/logical/TableSubQueryNode.java | 9 +-
.../engine/planner/physical/SeqScanExec.java | 3 +-
.../apache/tajo/engine/utils/SchemaUtil.java | 9 +
.../apache/tajo/engine/eval/ExprTestBase.java | 3 +-
25 files changed, 1443 insertions(+), 1013 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 8ecc7a2..af59b4c 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -449,6 +449,8 @@ Release 0.8.0 - unreleased
TASKS
+ TAJO-530: Fix warnings in tajo-catalog. (jaehwa)
+
TAJO-532: Fix warnings in tajo-common. (jinho)
TAJO-520: Move tajo-core-storage to tajo-storage. (jinho)
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
index 95bd83d..9f143ac 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
@@ -26,11 +26,7 @@ import org.apache.tajo.catalog.proto.CatalogProtos.SchemaProto;
import org.apache.tajo.catalog.proto.CatalogProtos.TableDescProto;
import org.apache.tajo.common.TajoDataTypes.DataType;
-import java.sql.Connection;
-import java.sql.ResultSet;
-import java.sql.Statement;
-import java.sql.PreparedStatement;
-import java.sql.Wrapper;
+import java.sql.*;
import java.util.Collection;
import static org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
@@ -157,21 +153,55 @@ public class CatalogUtil {
return sb.toString();
}
- public static void closeSQLWrapper(Wrapper... wrapper) {
- if(wrapper == null) return;
-
- for(Wrapper w : wrapper){
- try{
- if(w instanceof Statement){
- ((Statement)w).close();
- } else if(w instanceof PreparedStatement){
- ((PreparedStatement)w).close();
- } else if(w instanceof ResultSet){
- ((ResultSet)w).close();
- } else if(w instanceof Connection){
- ((Connection)w).close();
- }
- } catch (Exception e){}
+ public static void closeQuietly(Connection conn) {
+ try {
+ if (conn != null)
+ conn.close();
+ } catch (SQLException se) {
+ }
+ }
+
+ public static void closeQuietly(Statement stmt) {
+ try {
+ if (stmt != null)
+ stmt.close();
+ } catch (SQLException se) {
+ }
+ }
+
+ public static void closeQuietly(ResultSet res) {
+ try {
+ if (res != null)
+ res.close();
+ } catch (SQLException se) {
+ }
+ }
+
+ public static void closeQuietly(Connection conn, Statement stmt) {
+ try {
+ closeQuietly(stmt);
+ } finally {
+ closeQuietly(conn);
+ }
+ }
+
+ public static void closeQuietly(Connection conn, ResultSet res) {
+ try {
+ closeQuietly(res);
+ } finally {
+ closeQuietly(conn);
+ }
+ }
+
+ public static void closeQuietly(Connection conn, Statement stmt, ResultSet res) {
+ try {
+ closeQuietly(res);
+ } finally {
+ try {
+ closeQuietly(stmt);
+ } finally {
+ closeQuietly(conn);
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/FunctionDesc.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/FunctionDesc.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/FunctionDesc.java
index 084d97c..0ef8497 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/FunctionDesc.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/FunctionDesc.java
@@ -208,13 +208,17 @@ public class FunctionDesc implements ProtoObject<FunctionDescProto>, Cloneable,
}
public static String dataTypesToStr(List<DataType> parameterTypesList) {
- String result = "";
- String prefix = "";
- for(DataType eachType: parameterTypesList) {
- result += prefix + eachType.getType().toString();
- prefix = ",";
+ StringBuilder result = new StringBuilder();
+ for (int i = 0; i < parameterTypesList.size(); i++) {
+ DataType eachType = parameterTypesList.get(i);
+
+ if (i > 0) {
+ result.append(",");
+ }
+ result.append(eachType.getType().toString());
+
}
- return result;
+ return result.toString();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Options.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Options.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Options.java
index e770b11..c6f9a77 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Options.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Options.java
@@ -18,6 +18,7 @@
package org.apache.tajo.catalog;
+import com.google.common.base.Objects;
import com.google.gson.annotations.Expose;
import org.apache.tajo.catalog.json.CatalogGsonHelper;
import org.apache.tajo.catalog.proto.CatalogProtos.KeyValueProto;
@@ -98,8 +99,14 @@ public class Options implements ProtoObject<KeyValueSetProto>, Cloneable, GsonOb
public String delete(String key) {
return keyVals.remove(key);
}
-
- @Override
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(keyVals);
+
+ }
+
+ @Override
public boolean equals(Object object) {
if(object instanceof Options) {
Options other = (Options)object;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Schema.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Schema.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Schema.java
index f253151..5f1e8d7 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Schema.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Schema.java
@@ -18,12 +18,14 @@
package org.apache.tajo.catalog;
+import com.google.common.base.Objects;
import com.google.common.collect.ImmutableList;
import com.google.gson.annotations.Expose;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.tajo.catalog.exception.AlreadyExistsFieldException;
import org.apache.tajo.catalog.json.CatalogGsonHelper;
+import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.proto.CatalogProtos.ColumnProto;
import org.apache.tajo.catalog.proto.CatalogProtos.SchemaProto;
import org.apache.tajo.common.ProtoObject;
@@ -279,6 +281,11 @@ public class Schema implements ProtoObject<SchemaProto>, Cloneable, GsonObject {
}
@Override
+ public int hashCode() {
+ return Objects.hashCode(fields, fieldsByQualifiedName, fieldsByName);
+ }
+
+ @Override
public boolean equals(Object o) {
if (o instanceof Schema) {
Schema other = (Schema) o;
@@ -288,8 +295,15 @@ public class Schema implements ProtoObject<SchemaProto>, Cloneable, GsonObject {
}
@Override
- public Object clone() {
- Schema schema = new Schema(toArray());
+ public Object clone() throws CloneNotSupportedException {
+ Schema schema = null;
+
+ schema = (Schema) super.clone();
+ schema.builder = CatalogProtos.SchemaProto.newBuilder();
+ schema.init();
+ for(Column column: this.fields) {
+ schema.addColumn(column);
+ }
return schema;
}
@@ -311,7 +325,7 @@ public class Schema implements ProtoObject<SchemaProto>, Cloneable, GsonObject {
for(Column col : fields) {
sb.append(col);
if (i < fields.size() - 1) {
- sb.append(", ");
+ sb.append(",");
}
i++;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/SortSpec.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/SortSpec.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/SortSpec.java
index 5cc0de1..3ef73d5 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/SortSpec.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/SortSpec.java
@@ -18,6 +18,7 @@
package org.apache.tajo.catalog;
+import com.google.common.base.Objects;
import com.google.gson.annotations.Expose;
import org.apache.tajo.catalog.json.CatalogGsonHelper;
import org.apache.tajo.common.ProtoObject;
@@ -85,6 +86,11 @@ public class SortSpec implements Cloneable, GsonObject, ProtoObject<SortSpecProt
}
@Override
+ public int hashCode() {
+ return Objects.hashCode(Objects.hashCode(sortKey), ascending, nullFirst);
+ }
+
+ @Override
public boolean equals(Object object) {
if (object instanceof SortSpec) {
SortSpec other = (SortSpec) object;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/TableDesc.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/TableDesc.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/TableDesc.java
index 1b0a9fe..153d5f0 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/TableDesc.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/TableDesc.java
@@ -18,6 +18,7 @@
package org.apache.tajo.catalog;
+import com.google.common.base.Objects;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.annotations.Expose;
@@ -140,6 +141,10 @@ public class TableDesc implements ProtoObject<TableDescProto>, GsonObject, Clone
this.partitionMethodDesc = partitionMethodDesc;
}
+ public int hashCode() {
+ return Objects.hashCode(tableName, schema, meta, uri, stats, partitionMethodDesc);
+ }
+
public boolean equals(Object object) {
if(object instanceof TableDesc) {
TableDesc other = (TableDesc) object;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/json/TableMetaAdapter.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/json/TableMetaAdapter.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/json/TableMetaAdapter.java
index ce42bea..b3f788e 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/json/TableMetaAdapter.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/json/TableMetaAdapter.java
@@ -18,9 +18,9 @@
package org.apache.tajo.catalog.json;
+import com.google.common.base.Preconditions;
import com.google.gson.*;
import org.apache.tajo.catalog.Options;
-import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.json.GsonSerDerAdapter;
@@ -32,13 +32,17 @@ public class TableMetaAdapter implements GsonSerDerAdapter<TableMeta> {
@Override
public TableMeta deserialize(JsonElement json, Type typeOfT,
JsonDeserializationContext context) throws JsonParseException {
+ Preconditions.checkNotNull(json);
JsonObject jsonObject = json.getAsJsonObject();
CatalogProtos.StoreType type = CatalogProtos.StoreType.valueOf(jsonObject.get("store").getAsString());
Options options = null;
- if (jsonObject.has("options")) {
+ if (jsonObject.get("options") != null) {
options = context.deserialize(jsonObject.get("options"), Options.class);
+ } else {
+ throw new JsonParseException("Options not found in json");
+
}
return new TableMeta(type, options);
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/partition/PartitionDesc.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/partition/PartitionDesc.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/partition/PartitionDesc.java
index f7c1342..e89ee72 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/partition/PartitionDesc.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/partition/PartitionDesc.java
@@ -18,6 +18,7 @@
package org.apache.tajo.catalog.partition;
+import com.google.common.base.Objects;
import com.google.gson.annotations.Expose;
import org.apache.tajo.catalog.json.CatalogGsonHelper;
import org.apache.tajo.catalog.proto.CatalogProtos;
@@ -104,6 +105,10 @@ public class PartitionDesc implements ProtoObject<CatalogProtos.PartitionDescPro
this.path = path;
}
+ public int hashCode() {
+ return Objects.hashCode(tableId, partitionName, ordinalPosition, partitionValue, path);
+ }
+
public boolean equals(Object o) {
if (o instanceof PartitionDesc) {
PartitionDesc another = (PartitionDesc) o;
@@ -160,4 +165,17 @@ public class PartitionDesc implements ProtoObject<CatalogProtos.PartitionDescPro
public static PartitionDesc fromJson(String strVal) {
return strVal != null ? CatalogGsonHelper.fromJson(strVal, PartitionDesc.class) : null;
}
+
+ public Object clone() throws CloneNotSupportedException {
+ PartitionDesc desc = (PartitionDesc) super.clone();
+ desc.builder = CatalogProtos.PartitionDescProto.newBuilder();
+ desc.tableId = tableId;
+ desc.partitionName = partitionName;
+ desc.ordinalPosition = ordinalPosition;
+ desc.partitionValue = partitionValue;
+ desc.path = path;
+
+ return desc;
+ }
+
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/partition/PartitionMethodDesc.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/partition/PartitionMethodDesc.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/partition/PartitionMethodDesc.java
index d4a2c3e..4a8123a 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/partition/PartitionMethodDesc.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/partition/PartitionMethodDesc.java
@@ -18,6 +18,7 @@
package org.apache.tajo.catalog.partition;
+import com.google.common.base.Objects;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.annotations.Expose;
@@ -102,6 +103,11 @@ public class PartitionMethodDesc implements ProtoObject<CatalogProtos.PartitionM
}
@Override
+ public int hashCode() {
+ return Objects.hashCode(tableId, partitionType, expression, expressionSchema);
+ }
+
+ @Override
public CatalogProtos.PartitionMethodProto getProto() {
if(builder == null) {
builder = CatalogProtos.PartitionMethodProto.newBuilder();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/StatSet.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/StatSet.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/StatSet.java
index f8d24b0..9c95a1d 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/StatSet.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/StatSet.java
@@ -18,9 +18,11 @@
package org.apache.tajo.catalog.statistics;
+import com.google.common.base.Objects;
import com.google.common.collect.Maps;
import com.google.gson.annotations.Expose;
import org.apache.tajo.catalog.json.CatalogGsonHelper;
+import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.proto.CatalogProtos.StatProto;
import org.apache.tajo.catalog.proto.CatalogProtos.StatSetProto;
import org.apache.tajo.catalog.proto.CatalogProtos.StatSetProtoOrBuilder;
@@ -67,7 +69,11 @@ public class StatSet implements ProtoObject<StatSetProto>, Cloneable {
initStats();
return stats.values();
}
-
+
+ public int hashCode() {
+ return Objects.hashCode(viaProto, stats);
+ }
+
public boolean equals(Object obj) {
if (obj instanceof StatSet) {
StatSet other = (StatSet) obj;
@@ -83,7 +89,18 @@ public class StatSet implements ProtoObject<StatSetProto>, Cloneable {
}
public Object clone() throws CloneNotSupportedException {
- return new StatSet(this.getProto());
+ StatSet statSet = (StatSet) super.clone();
+ statSet.builder = CatalogProtos.StatSetProto.newBuilder();
+ statSet.viaProto = viaProto;
+
+ statSet.stats = Maps.newHashMap();
+ for (Entry<StatType, Stat> entry : stats.entrySet()) {
+ StatType type = (StatType)entry.getKey();
+ Stat stat = (Stat)entry.getValue().clone();
+ statSet.stats.put(type, stat);
+ }
+
+ return statSet;
}
private void initStats() {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java
index f796fff..bc43f2c 100644
--- a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java
+++ b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java
@@ -31,6 +31,7 @@ import org.apache.hcatalog.data.schema.HCatFieldSchema;
import org.apache.hcatalog.data.schema.HCatSchema;
import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.exception.CatalogException;
import org.apache.tajo.catalog.partition.PartitionMethodDesc;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.statistics.TableStats;
@@ -38,7 +39,10 @@ import org.apache.tajo.common.TajoDataTypes;
import org.apache.tajo.exception.InternalException;
import java.io.IOException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Properties;
import static org.apache.tajo.catalog.proto.CatalogProtos.PartitionType;
@@ -47,7 +51,7 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
protected final Log LOG = LogFactory.getLog(getClass());
protected Configuration conf;
- private static final int CLIENT_POOL_SIZE = 5;
+ private static final int CLIENT_POOL_SIZE = 2;
private final HCatalogStoreClientPool clientPool = new HCatalogStoreClientPool(0);
public HCatalogStore(final Configuration conf)
@@ -61,7 +65,7 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
}
@Override
- public boolean existTable(final String name) throws IOException {
+ public boolean existTable(final String name) throws CatalogException {
boolean exist = false;
String dbName = null, tableName = null;
@@ -73,8 +77,8 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
tablePair = HCatUtil.getDbAndTableName(name);
dbName = tablePair.first;
tableName = tablePair.second;
- } catch (IOException ioe) {
- throw new InternalException("Table name is wrong.", ioe);
+ } catch (Exception ioe) {
+ throw new CatalogException("Table name is wrong.", ioe);
}
// get table
@@ -88,7 +92,7 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
} catch (NoSuchObjectException nsoe) {
exist = false;
} catch (Exception e) {
- throw new IOException(e);
+ throw new CatalogException(e);
}
} finally {
client.release();
@@ -98,7 +102,7 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
}
@Override
- public final CatalogProtos.TableDescProto getTable(final String name) throws IOException {
+ public final CatalogProtos.TableDescProto getTable(final String name) throws CatalogException {
String dbName = null, tableName = null;
Pair<String, String> tablePair = null;
org.apache.hadoop.hive.ql.metadata.Table table = null;
@@ -115,8 +119,8 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
tablePair = HCatUtil.getDbAndTableName(name);
dbName = tablePair.first;
tableName = tablePair.second;
- } catch (IOException ioe) {
- throw new InternalException("Table name is wrong.", ioe);
+ } catch (Exception ioe) {
+ throw new CatalogException("Table name is wrong.", ioe);
}
//////////////////////////////////
@@ -129,14 +133,20 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
table = HCatUtil.getTable(client.getHiveClient(), dbName, tableName);
path = table.getPath();
} catch (NoSuchObjectException nsoe) {
- throw new InternalException("Table not found. - tableName:" + name, nsoe);
+ throw new CatalogException("Table not found. - tableName:" + name, nsoe);
} catch (Exception e) {
- throw new IOException(e);
+ throw new CatalogException(e);
}
// convert hcatalog field schema into tajo field schema.
schema = new org.apache.tajo.catalog.Schema();
- HCatSchema tableSchema = HCatUtil.getTableSchemaWithPtnCols(table);
+ HCatSchema tableSchema = null;
+
+ try {
+ tableSchema = HCatUtil.getTableSchemaWithPtnCols(table);
+ } catch (IOException ioe) {
+ throw new CatalogException("Fail to get table schema. - tableName:" + name, ioe);
+ }
List<HCatFieldSchema> fieldSchemaList = tableSchema.getFields();
boolean isPartitionKey = false;
for (HCatFieldSchema eachField : fieldSchemaList) {
@@ -160,9 +170,8 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
// validate field schema.
try {
HCatalogUtil.validateHCatTableAndTajoSchema(tableSchema);
- } catch (IOException e) {
- throw new InternalException(
- "HCatalog cannot support schema. - schema:" + tableSchema.toString(), e);
+ } catch (Exception e) {
+ throw new CatalogException("HCatalog cannot support schema. - schema:" + tableSchema.toString(), e);
}
stats = new TableStats();
@@ -181,8 +190,7 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
// set file output format
fileOutputformat = properties.getProperty("file.outputformat");
- storeType = CatalogUtil.getStoreType(HCatalogUtil.getStoreType(fileOutputformat,
- fieldDelimiter));
+ storeType = CatalogUtil.getStoreType(HCatalogUtil.getStoreType(fileOutputformat));
if (storeType.equals(CatalogProtos.StoreType.CSV) ) {
options.put(CVSFILE_DELIMITER, fieldDelimiter);
@@ -193,9 +201,13 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
if(properties.getProperty("totalSize") != null) {
totalSize = new Long(properties.getProperty("totalSize"));
} else {
- FileSystem fs = path.getFileSystem(conf);
- if (fs.exists(path)) {
- totalSize = fs.getContentSummary(path).getLength();
+ try {
+ FileSystem fs = path.getFileSystem(conf);
+ if (fs.exists(path)) {
+ totalSize = fs.getContentSummary(path).getLength();
+ }
+ } catch (IOException ioe) {
+ throw new CatalogException("Fail to get path. - path:" + path.toString(), ioe);
}
}
stats.setNumBytes(totalSize);
@@ -249,26 +261,23 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
}
@Override
- public final List<String> getAllTableNames() throws IOException {
+ public final List<String> getAllTableNames() throws CatalogException {
List<String> dbs = null;
List<String> tables = null;
List<String> allTables = new ArrayList<String>();
HCatalogStoreClientPool.HCatalogStoreClient client = null;
try {
- try {
- client = clientPool.getClient();
- dbs = client.getHiveClient().getAllDatabases();
- for(String eachDB: dbs) {
- tables = client.getHiveClient().getAllTables(eachDB);
- for(String eachTable: tables) {
- allTables.add(eachDB + "." + eachTable);
- }
+ client = clientPool.getClient();
+ dbs = client.getHiveClient().getAllDatabases();
+ for(String eachDB: dbs) {
+ tables = client.getHiveClient().getAllTables(eachDB);
+ for(String eachTable: tables) {
+ allTables.add(eachDB + "." + eachTable);
}
- } catch (Exception e) {
- throw new IOException(e);
}
-
+ } catch (MetaException e) {
+ throw new CatalogException(e);
} finally {
client.release();
}
@@ -276,7 +285,7 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
}
@Override
- public final void addTable(final CatalogProtos.TableDescProto tableDesc) throws IOException {
+ public final void addTable(final CatalogProtos.TableDescProto tableDesc) throws CatalogException {
String dbName = null, tableName = null;
Pair<String, String> tablePair = null;
HCatalogStoreClientPool.HCatalogStoreClient client = null;
@@ -286,88 +295,88 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
tablePair = HCatUtil.getDbAndTableName(tableDesc.getId());
dbName = tablePair.first;
tableName = tablePair.second;
- } catch (IOException ioe) {
- throw new InternalException("Table name is wrong.", ioe);
+ } catch (Exception ioe) {
+ throw new CatalogException("Table name is wrong.", ioe);
}
try {
- try {
- client = clientPool.getClient();
+ client = clientPool.getClient();
- org.apache.hadoop.hive.metastore.api.Table table = new org.apache.hadoop.hive.metastore.api.Table();
+ org.apache.hadoop.hive.metastore.api.Table table = new org.apache.hadoop.hive.metastore.api.Table();
- table.setDbName(dbName);
- table.setTableName(tableName);
- // TODO: set owner
- //table.setOwner();
+ table.setDbName(dbName);
+ table.setTableName(tableName);
+ // TODO: set owner
+ //table.setOwner();
- StorageDescriptor sd = new StorageDescriptor();
+ StorageDescriptor sd = new StorageDescriptor();
- // if tajo set location method, thrift client make exception as follows:
- // Caused by: MetaException(message:java.lang.NullPointerException)
- // If you want to modify table path, you have to modify on Hive cli.
- //sd.setLocation(tableDesc.getPath().toString());
+ // if tajo set location method, thrift client make exception as follows:
+ // Caused by: MetaException(message:java.lang.NullPointerException)
+ // If you want to modify table path, you have to modify on Hive cli.
+ //sd.setLocation(tableDesc.getPath().toString());
- // set column information
- ArrayList<FieldSchema> cols = new ArrayList<FieldSchema>(tableDesc.getSchema().getFieldsCount());
- for (CatalogProtos.ColumnProto col : tableDesc.getSchema().getFieldsList()) {
- cols.add(new FieldSchema(
- col.getColumnName(),
- HCatalogUtil.getHiveFieldType(col.getDataType().getType().name()),
- ""));
- }
- sd.setCols(cols);
-
- sd.setCompressed(false);
- if (tableDesc.getMeta().hasParams()) {
- for (CatalogProtos.KeyValueProto entry: tableDesc.getMeta().getParams().getKeyvalList()) {
- if (entry.getKey().equals("compression.codec")) {
- sd.setCompressed(true);
- } else if (entry.getKey().equals(CVSFILE_DELIMITER)) {
- sd.getSerdeInfo().getParameters().put(serdeConstants.FIELD_DELIM, entry.getValue());
- }
+ // set column information
+ ArrayList<FieldSchema> cols = new ArrayList<FieldSchema>(tableDesc.getSchema().getFieldsCount());
+ for (CatalogProtos.ColumnProto col : tableDesc.getSchema().getFieldsList()) {
+ cols.add(new FieldSchema(
+ col.getColumnName(),
+ HCatalogUtil.getHiveFieldType(col.getDataType().getType().name()),
+ ""));
+ }
+ sd.setCols(cols);
+
+ sd.setCompressed(false);
+ if (tableDesc.getMeta().hasParams()) {
+ for (CatalogProtos.KeyValueProto entry: tableDesc.getMeta().getParams().getKeyvalList()) {
+ if (entry.getKey().equals("compression.codec")) {
+ sd.setCompressed(true);
+ } else if (entry.getKey().equals(CVSFILE_DELIMITER)) {
+ sd.getSerdeInfo().getParameters().put(serdeConstants.FIELD_DELIM, entry.getValue());
}
}
+ }
- sd.setParameters(new HashMap<String, String>());
- sd.setSerdeInfo(new SerDeInfo());
- sd.getSerdeInfo().setName(table.getTableName());
+ sd.setParameters(new HashMap<String, String>());
+ sd.setSerdeInfo(new SerDeInfo());
+ sd.getSerdeInfo().setName(table.getTableName());
- if(tableDesc.getMeta().getStoreType().equals(CatalogProtos.StoreType.RCFILE)) {
- sd.getSerdeInfo().setSerializationLib(org.apache.hadoop.hive.serde2.columnar.ColumnarSerDe.class.getName());
- } else {
- sd.getSerdeInfo().setSerializationLib(org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe.class.getName());
- }
+ if(tableDesc.getMeta().getStoreType().equals(CatalogProtos.StoreType.RCFILE)) {
+ sd.getSerdeInfo().setSerializationLib(org.apache.hadoop.hive.serde2.columnar.ColumnarSerDe.class.getName());
+ } else {
+ sd.getSerdeInfo().setSerializationLib(org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe.class.getName());
+ }
- sd.getSerdeInfo().setParameters(new HashMap<String, String>());
+ sd.getSerdeInfo().setParameters(new HashMap<String, String>());
// sd.getSerdeInfo().getParameters().put(serdeConstants.SERIALIZATION_FORMAT, "1");
- if(tableDesc.getMeta().getStoreType().equals(CatalogProtos.StoreType.RCFILE)) {
- sd.setInputFormat(org.apache.hadoop.hive.ql.io.RCFileInputFormat.class.getName());
- } else {
- sd.setInputFormat(org.apache.hadoop.mapred.TextInputFormat.class.getName());
- }
+ if(tableDesc.getMeta().getStoreType().equals(CatalogProtos.StoreType.RCFILE)) {
+ sd.setInputFormat(org.apache.hadoop.hive.ql.io.RCFileInputFormat.class.getName());
+ } else {
+ sd.setInputFormat(org.apache.hadoop.mapred.TextInputFormat.class.getName());
+ }
- if(tableDesc.getMeta().getStoreType().equals(CatalogProtos.StoreType.RCFILE)) {
- sd.setOutputFormat(org.apache.hadoop.hive.ql.io.RCFileOutputFormat.class.getName());
- } else {
- sd.setOutputFormat(org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat.class.getName());
- }
+ if(tableDesc.getMeta().getStoreType().equals(CatalogProtos.StoreType.RCFILE)) {
+ sd.setOutputFormat(org.apache.hadoop.hive.ql.io.RCFileOutputFormat.class.getName());
+ } else {
+ sd.setOutputFormat(org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat.class.getName());
+ }
- sd.setSortCols(new ArrayList<Order>());
+ sd.setSortCols(new ArrayList<Order>());
- table.setSd(sd);
- client.getHiveClient().createTable(table);
- } catch (Exception e) {
- throw new IOException(e);
- }
+ table.setSd(sd);
+ client.getHiveClient().createTable(table);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new CatalogException(e);
} finally {
client.release();
}
}
@Override
- public final void deleteTable(final String name) throws IOException {
+ public final void deleteTable(final String name) throws CatalogException {
String dbName = null, tableName = null;
Pair<String, String> tablePair = null;
HCatalogStoreClientPool.HCatalogStoreClient client = null;
@@ -377,8 +386,8 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
tablePair = HCatUtil.getDbAndTableName(name);
dbName = tablePair.first;
tableName = tablePair.second;
- } catch (IOException ioe) {
- throw new InternalException("Table name is wrong.", ioe);
+ } catch (Exception ioe) {
+ throw new CatalogException("Table name is wrong.", ioe);
}
try {
@@ -386,120 +395,119 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
client.getHiveClient().dropTable(dbName, tableName, false, false);
} catch (NoSuchObjectException nsoe) {
} catch (Exception e) {
- throw new IOException(e);
+ throw new CatalogException(e);
} finally {
client.release();
}
}
@Override
- public void addPartitionMethod(CatalogProtos.PartitionMethodProto partitionMethodProto) throws IOException {
+ public void addPartitionMethod(CatalogProtos.PartitionMethodProto partitionMethodProto) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public CatalogProtos.PartitionMethodProto getPartitionMethod(String tableName) throws IOException {
+ public CatalogProtos.PartitionMethodProto getPartitionMethod(String tableName) throws CatalogException {
return null; // TODO - not implemented yet
}
@Override
- public boolean existPartitionMethod(String tableName) throws IOException {
+ public boolean existPartitionMethod(String tableName) throws CatalogException {
return false; // TODO - not implemented yet
}
@Override
- public void delPartitionMethod(String tableName) throws IOException {
+ public void delPartitionMethod(String tableName) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public void addPartitions(CatalogProtos.PartitionsProto partitionsProto) throws IOException {
+ public void addPartitions(CatalogProtos.PartitionsProto partitionsProto) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public void addPartition(CatalogProtos.PartitionDescProto partitionDescProto) throws IOException {
+ public void addPartition(CatalogProtos.PartitionDescProto partitionDescProto) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public CatalogProtos.PartitionsProto getPartitions(String tableName) throws IOException {
+ public CatalogProtos.PartitionsProto getPartitions(String tableName) throws CatalogException {
return null; // TODO - not implemented yet
}
@Override
- public CatalogProtos.PartitionDescProto getPartition(String partitionName) throws IOException {
+ public CatalogProtos.PartitionDescProto getPartition(String partitionName) throws CatalogException {
return null; // TODO - not implemented yet
}
@Override
- public void delPartition(String partitionName) throws IOException {
+ public void delPartition(String partitionName) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public void delPartitions(String tableName) throws IOException {
+ public void delPartitions(String tableName) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public final void addFunction(final FunctionDesc func) throws IOException {
+ public final void addFunction(final FunctionDesc func) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public final void deleteFunction(final FunctionDesc func) throws IOException {
+ public final void deleteFunction(final FunctionDesc func) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public final void existFunction(final FunctionDesc func) throws IOException {
+ public final void existFunction(final FunctionDesc func) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public final List<String> getAllFunctionNames() throws IOException {
+ public final List<String> getAllFunctionNames() throws CatalogException {
// TODO - not implemented yet
return null;
}
@Override
- public void delIndex(String indexName) throws IOException {
+ public void delIndex(String indexName) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public boolean existIndex(String indexName) throws IOException {
+ public boolean existIndex(String indexName) throws CatalogException {
// TODO - not implemented yet
return false;
}
@Override
- public CatalogProtos.IndexDescProto[] getIndexes(String tableName) throws IOException {
+ public CatalogProtos.IndexDescProto[] getIndexes(String tableName) throws CatalogException {
// TODO - not implemented yet
return null;
}
@Override
- public void addIndex(CatalogProtos.IndexDescProto proto) throws IOException {
+ public void addIndex(CatalogProtos.IndexDescProto proto) throws CatalogException {
// TODO - not implemented yet
}
@Override
- public CatalogProtos.IndexDescProto getIndex(String indexName) throws IOException {
+ public CatalogProtos.IndexDescProto getIndex(String indexName) throws CatalogException {
// TODO - not implemented yet
return null;
}
@Override
- public CatalogProtos.IndexDescProto getIndex(String tableName, String columnName)
- throws IOException {
+ public CatalogProtos.IndexDescProto getIndex(String tableName, String columnName) throws CatalogException {
// TODO - not implemented yet
return null;
}
@Override
- public boolean existIndex(String tableName, String columnName) {
+ public boolean existIndex(String tableName, String columnName) throws CatalogException{
// TODO - not implemented yet
return false;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStoreClientPool.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStoreClientPool.java b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStoreClientPool.java
index 263e1dd..0a6c9b9 100644
--- a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStoreClientPool.java
+++ b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStoreClientPool.java
@@ -20,6 +20,7 @@ import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
import org.apache.log4j.Logger;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
/**
* Manages a pool of HiveMetaStoreClient connections. If the connection pool is empty
@@ -29,7 +30,7 @@ public class HCatalogStoreClientPool {
private static final Logger LOG = Logger.getLogger(HCatalogStoreClientPool.class);
private final ConcurrentLinkedQueue<HCatalogStoreClient> clientPool =
new ConcurrentLinkedQueue<HCatalogStoreClient>();
- private Boolean poolClosed = false;
+ private AtomicBoolean poolClosed = new AtomicBoolean(false);
private final HiveConf hiveConf;
/**
@@ -42,7 +43,9 @@ public class HCatalogStoreClientPool {
private HCatalogStoreClient(HiveConf hiveConf) {
try {
- LOG.debug("Creating MetaStoreClient. Pool Size = " + clientPool.size());
+
+ LOG.info("Creating MetaStoreClient. Pool Size = " + clientPool.size());
+
this.hiveClient = new HiveMetaStoreClient(hiveConf);
} catch (Exception e) {
// Turn in to an unchecked exception
@@ -69,15 +72,15 @@ public class HCatalogStoreClientPool {
// This lock is needed to ensure proper behavior when a thread reads poolClosed
// is false, but a call to pool.close() comes in immediately afterward.
synchronized (poolClosed) {
- if (poolClosed) {
+// if (poolClosed.get()) {
hiveClient.close();
- } else {
+// } else {
// TODO: Currently the pool does not work properly because we cannot
// reuse MetastoreClient connections. No reason to add this client back
// to the pool. See HIVE-5181.
- // clientPool.add(this);
- hiveClient.close();
- }
+// clientPool.add(this);
+// hiveClient.close();
+// }
}
}
@@ -122,11 +125,6 @@ public class HCatalogStoreClientPool {
// The pool was empty so create a new client and return that.
if (client == null) {
client = new HCatalogStoreClient(hiveConf);
- } else {
- // TODO: Due to Hive Metastore bugs, there is leftover state from previous client
- // connections so we are unable to reuse the same connection. For now simply
- // reconnect each time. One possible culprit is HIVE-5181.
- client = new HCatalogStoreClient(hiveConf);
}
client.markInUse();
return client;
@@ -139,10 +137,10 @@ public class HCatalogStoreClientPool {
public void close() {
// Ensure no more items get added to the pool once close is called.
synchronized (poolClosed) {
- if (poolClosed) {
+ if (poolClosed.get()) {
return;
}
- poolClosed = true;
+ poolClosed.set(true);
}
HCatalogStoreClient client = null;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogUtil.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogUtil.java b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogUtil.java
index 8310873..f2764fb 100644
--- a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogUtil.java
+++ b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogUtil.java
@@ -17,6 +17,7 @@
*/
package org.apache.tajo.catalog.store;
+import com.google.common.base.Preconditions;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat;
@@ -25,23 +26,20 @@ import org.apache.hadoop.hive.serde.serdeConstants;
import org.apache.hcatalog.common.HCatException;
import org.apache.hcatalog.data.schema.HCatFieldSchema;
import org.apache.hcatalog.data.schema.HCatSchema;
+import org.apache.tajo.catalog.exception.CatalogException;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.common.TajoDataTypes;
-import org.apache.tajo.exception.InternalException;
-
-import java.io.IOException;
public class HCatalogUtil {
protected final Log LOG = LogFactory.getLog(getClass());
- public static void validateHCatTableAndTajoSchema(HCatSchema tblSchema) throws InternalException {
+ public static void validateHCatTableAndTajoSchema(HCatSchema tblSchema) throws CatalogException {
for (HCatFieldSchema hcatField : tblSchema.getFields()) {
validateHCatFieldAndTajoSchema(hcatField);
}
}
- private static void validateHCatFieldAndTajoSchema(HCatFieldSchema fieldSchema) throws
- InternalException {
+ private static void validateHCatFieldAndTajoSchema(HCatFieldSchema fieldSchema) throws CatalogException {
try {
HCatFieldSchema.Type fieldType = fieldSchema.getType();
switch (fieldType) {
@@ -53,15 +51,14 @@ public class HCatalogUtil {
throw new HCatException("Tajo cannot support map field type.");
}
} catch (HCatException e) {
- throw new InternalException("incompatible hcatalog types when assigning to tajo type. - " +
- "HCatFieldSchema:" + fieldSchema, e);
+ throw new CatalogException("incompatible hcatalog types when assigning to tajo type. - " +
+ "HCatFieldSchema:" + fieldSchema);
}
}
- public static TajoDataTypes.Type getTajoFieldType(String fieldType) throws IOException {
- if(fieldType == null) {
- throw new InternalException("Hive field type is null.");
- }
+ public static TajoDataTypes.Type getTajoFieldType(String fieldType) {
+ Preconditions.checkNotNull(fieldType);
+
String typeStr = null;
if(fieldType.equalsIgnoreCase(serdeConstants.INT_TYPE_NAME))
@@ -86,14 +83,12 @@ public class HCatalogUtil {
try {
return Enum.valueOf(TajoDataTypes.Type.class, typeStr);
} catch (IllegalArgumentException iae) {
- throw new InternalException("Cannot find a matched type aginst from '" + typeStr + "'");
+ throw new CatalogException("Cannot find a matched type aginst from '" + typeStr + "'");
}
}
- public static String getHiveFieldType(String fieldType) throws IOException {
- if(fieldType == null) {
- throw new InternalException("Tajo field type is null.");
- }
+ public static String getHiveFieldType(String fieldType) {
+ Preconditions.checkNotNull(fieldType);
String typeStr = null;
if(fieldType.equalsIgnoreCase("INT4"))
@@ -118,14 +113,12 @@ public class HCatalogUtil {
return typeStr;
}
- public static String getStoreType(String fileFormat, String delimiter) throws IOException{
- if(fileFormat == null) {
- throw new InternalException("Hive file output format is null.");
- }
+ public static String getStoreType(String fileFormat) {
+ Preconditions.checkNotNull(fileFormat);
String[] fileFormatArrary = fileFormat.split("\\.");
if(fileFormatArrary.length < 1) {
- throw new InternalException("Hive file output format is wrong. - file output format:" + fileFormat);
+ throw new CatalogException("Hive file output format is wrong. - file output format:" + fileFormat);
}
String inputFormatClass = fileFormatArrary[fileFormatArrary.length-1];
@@ -135,7 +128,7 @@ public class HCatalogUtil {
} else if(inputFormatClass.equals(RCFileOutputFormat.class.getSimpleName())) {
return CatalogProtos.StoreType.RCFILE.name();
} else {
- throw new InternalException("Not supported file output format. - file output format:" + fileFormat);
+ throw new CatalogException("Not supported file output format. - file output format:" + fileFormat);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/413a4b3f/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
index cf13a9d..dc70305 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
@@ -43,10 +43,7 @@ import org.apache.tajo.util.TUtil;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.net.InetSocketAddress;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -103,7 +100,12 @@ public class CatalogServer extends AbstractService {
Constructor<?> cons;
try {
- this.conf = (TajoConf) conf;
+ if (conf instanceof TajoConf) {
+ this.conf = (TajoConf) conf;
+ } else {
+ throw new CatalogException();
+ }
+
Class<?> storeClass = this.conf.getClass(CatalogConstants.STORE_CLASS, DerbyStore.class);
LOG.info("Catalog Store Class: " + storeClass.getCanonicalName());
@@ -201,9 +203,9 @@ public class CatalogServer extends AbstractService {
}
return store.getTable(tableId);
- } catch (IOException ioe) {
+ } catch (Exception e) {
// TODO - handle exception
- LOG.error(ioe);
+ LOG.error(e);
return null;
} finally {
rlock.unlock();
@@ -222,8 +224,9 @@ public class CatalogServer extends AbstractService {
builder.addTableName(iterator.next());
}
return builder.build();
- } catch (IOException ioe) {
+ } catch (Exception e) {
// TODO - handle exception
+ LOG.error(e);
return null;
}
}
@@ -250,8 +253,8 @@ public class CatalogServer extends AbstractService {
throw new AlreadyExistsTableException(proto.getId());
}
store.addTable(proto);
- } catch (IOException ioe) {
- LOG.error(ioe.getMessage(), ioe);
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
return BOOL_FALSE;
} finally {
wlock.unlock();
@@ -272,8 +275,8 @@ public class CatalogServer extends AbstractService {
throw new NoSuchTableException(tableId);
}
store.deleteTable(tableId);
- } catch (IOException ioe) {
- LOG.error(ioe.getMessage(), ioe);
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
return BOOL_FALSE;
} finally {
wlock.unlock();
@@ -292,7 +295,7 @@ public class CatalogServer extends AbstractService {
} else {
return BOOL_FALSE;
}
- } catch (IOException e) {
+ } catch (Exception e) {
LOG.error(e);
throw new ServiceException(e);
}
@@ -306,9 +309,9 @@ public class CatalogServer extends AbstractService {
try {
String tableId = name.getValue().toLowerCase();
return store.getPartitionMethod(tableId);
- } catch (IOException ioe) {
+ } catch (Exception e) {
// TODO - handle exception
- LOG.error(ioe);
+ LOG.error(e);
return null;
} finally {
rlock.unlock();
@@ -323,7 +326,7 @@ public class CatalogServer extends AbstractService {
String tableId = tableName.getValue().toLowerCase();
return BoolProto.newBuilder().setValue(
store.existPartitionMethod(tableId)).build();
- } catch (IOException e) {
+ } catch (Exception e) {
LOG.error(e);
return BoolProto.newBuilder().setValue(false).build();
} finally {
@@ -379,10 +382,10 @@ public class CatalogServer extends AbstractService {
throw new AlreadyExistsIndexException(indexDesc.getName());
}
store.addIndex(indexDesc);
- } catch (IOException ioe) {
- LOG.error("ERROR : cannot add index " + indexDesc.getName(), ioe);
+ } catch (Exception e) {
+ LOG.error("ERROR : cannot add index " + indexDesc.getName(), e);
LOG.error(indexDesc);
- throw new ServiceException(ioe);
+ throw new ServiceException(e);
} finally {
rlock.unlock();
}
@@ -398,7 +401,7 @@ public class CatalogServer extends AbstractService {
try {
return BoolProto.newBuilder().setValue(
store.existIndex(indexName.getValue())).build();
- } catch (IOException e) {
+ } catch (Exception e) {
LOG.error(e);
return BoolProto.newBuilder().setValue(false).build();
} finally {
@@ -415,7 +418,7 @@ public class CatalogServer extends AbstractService {
return BoolProto.newBuilder().setValue(
store.existIndex(request.getTableName(),
request.getColumnName())).build();
- } catch (IOException e) {
+ } catch (Exception e) {
LOG.error(e);
return BoolProto.newBuilder().setValue(false).build();
} finally {
@@ -433,8 +436,8 @@ public class CatalogServer extends AbstractService {
throw new NoSuchIndexException(indexName.getValue());
}
return store.getIndex(indexName.getValue());
- } catch (IOException ioe) {
- LOG.error("ERROR : cannot get index " + indexName, ioe);
+ } catch (Exception e) {
+ LOG.error("ERROR : cannot get index " + indexName, e);
return null;
} finally {
rlock.unlock();
@@ -452,9 +455,9 @@ public class CatalogServer extends AbstractService {
+ request.getColumnName());
}
return store.getIndex(request.getTableName(), request.getColumnName());
- } catch (IOException ioe) {
+ } catch (Exception e) {
LOG.error("ERROR : cannot get index " + request.getTableName() + "."
- + request.getColumnName(), ioe);
+ + request.getColumnName(), e);
return null;
} finally {
rlock.unlock();
@@ -470,7 +473,7 @@ public class CatalogServer extends AbstractService {
throw new NoSuchIndexException(indexName.getValue());
}
store.delIndex(indexName.getValue());
- } catch (IOException e) {
+ } catch (Exception e) {
LOG.error(e);
} finally {
wlock.unlock();
@@ -593,7 +596,7 @@ public class CatalogServer extends AbstractService {
}
}
- private static class FunctionSignature implements Comparable<FunctionSignature> {
+ private static class FunctionSignature {
private String signature;
private FunctionType type;
private DataType [] arguments;
@@ -637,37 +640,23 @@ public class CatalogServer extends AbstractService {
}
@Override
- public int hashCode() {
- return Objects.hashCode(signature, type, Objects.hashCode(arguments));
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ } else if (o == null || getClass() != o.getClass()) {
+ return false;
+ } else {
+ return (signature.equals(((FunctionSignature) o).signature)
+ && type.equals(((FunctionSignature) o).type)
+ && Arrays.equals(arguments, ((FunctionSignature) o).arguments));
+ }
}
@Override
- public int compareTo(FunctionSignature o) {
- int signatureCmp = signature.compareTo(o.signature);
- if (signatureCmp != 0) {
- return signatureCmp;
- }
-
- int typeCmp = type.compareTo(o.type);
- if (typeCmp != 0) {
- return typeCmp;
- }
-
- int min = Math.min(arguments.length, o.arguments.length);
- int argCmp = 0;
- for (int i = 0; i < min; i++) {
- if (arguments.length < min && o.arguments.length < min) {
- argCmp = arguments[i].getType().compareTo(o.arguments[i].getType());
-
- if (argCmp != 0) {
- return argCmp;
- }
- } else {
- argCmp = arguments.length - o.arguments.length;
- }
- }
- return argCmp;
+ public int hashCode() {
+ return Objects.hashCode(signature, type, Objects.hashCode(arguments));
}
+
}
public static void main(String[] args) throws Exception {