working version of new simple-connector interface

mitdbg · Jul 24, 2018 · e2dde6b · e2dde6b
1 parent e6e45f5
commit e2dde6b
Show file tree

Hide file tree

Showing 3 changed files with 28 additions and 13 deletions.
diff --git a/ddprofiler/src/main/java/sources/deprecated/BenchmarkingData.java b/ddprofiler/src/main/java/sources/deprecated/BenchmarkingData.java
@@ -8,6 +8,7 @@
 
 import au.com.bytecode.opencsv.CSVReader;
 
+@Deprecated
 public class BenchmarkingData {
 
     public List<Attribute> attributes;

diff --git a/ddprofiler/src/main/java/sources/implementations/CSVSource.java b/ddprofiler/src/main/java/sources/implementations/CSVSource.java
@@ -25,7 +25,6 @@
 import sources.config.SourceConfig;
 import sources.deprecated.Attribute;
 import sources.deprecated.Record;
-import sources.deprecated.TableInfo;
 
 public class CSVSource implements Source {
 
@@ -37,7 +36,8 @@ public class CSVSource implements Source {
     private CSVSourceConfig config;
     private boolean initialized = false;
     private CSVReader fileReader;
-    private TableInfo tableInfo;
+    // private TableInfo tableInfo;
+    private List<Attribute> attributes;
 
     // metrics
     private long lineCounter = 0;
@@ -48,10 +48,11 @@ public CSVSource() {
 
     }
 
-    public CSVSource(String path, String relationName) {
+    public CSVSource(String path, String relationName, SourceConfig config) {
 	this.tid = SourceUtils.computeTaskId(path, relationName);
 	this.path = path;
 	this.relationName = relationName;
+	this.config = (CSVSourceConfig) config;
     }
 
     @Override
@@ -98,7 +99,7 @@ public List<Source> processSource(SourceConfig config) {
 		String path = f.getParent() + File.separator;
 		String name = f.getName();
 		// Make the csv config specific to the relation
-		CSVSource task = new CSVSource(path, name);
+		CSVSource task = new CSVSource(path, name, config);
 		totalFiles++;
 		// c.submitTask(pt);
 		tasks.add(task);
@@ -116,9 +117,16 @@ public SourceType getSourceType() {
 
     @Override
     public List<Attribute> getAttributes() throws IOException, SQLException {
+	if (!initialized) {
+	    String path = this.path + this.relationName;
+	    char separator = this.config.getSeparator().charAt(0);
+	    fileReader = new CSVReader(new FileReader(path), separator);
+	    initialized = true;
+	}
 	// assume that the first row is the attributes;
 	if (lineCounter != 0) {
-	    return tableInfo.getTableAttributes();
+	    // return tableInfo.getTableAttributes();
+	    return attributes;
 	}
 	String[] attributes = fileReader.readNext();
 	lineCounter++;
@@ -128,6 +136,7 @@ public List<Attribute> getAttributes() throws IOException, SQLException {
 	    Attribute attr = new Attribute(attributes[i]);
 	    attrList.add(attr);
 	}
+	this.attributes = attrList;
 	return attrList;
     }
 

diff --git a/ddprofiler/src/main/java/sources/implementations/PostgresSource.java b/ddprofiler/src/main/java/sources/implementations/PostgresSource.java
@@ -30,7 +30,6 @@
 import sources.config.SourceConfig;
 import sources.deprecated.Attribute;
 import sources.deprecated.Record;
-import sources.deprecated.TableInfo;
 
 public class PostgresSource implements Source {
 
@@ -48,7 +47,7 @@ public class PostgresSource implements Source {
     private Statement theStatement;
     private ResultSet theRS;
 
-    private TableInfo tableInfo;
+    private List<Attribute> attributes;
 
     // Metrics
     private Counter error_records = Metrics.REG.counter((name(PostgresSource.class, "error", "records")));
@@ -61,8 +60,9 @@ public PostgresSource() {
 
     }
 
-    public PostgresSource(String relationName) {
+    public PostgresSource(String relationName, PostgresSourceConfig config) {
 	this.relationName = relationName;
+	this.config = config;
 	this.tid = SourceUtils.computeTaskId(this.config.getDatabase_name(), relationName);
     }
 
@@ -101,7 +101,7 @@ public List<Source> processSource(SourceConfig config) {
 	    PostgresSourceConfig relationPostgresSourceConfig = (PostgresSourceConfig) postgresConfig.selfCopy();
 	    relationPostgresSourceConfig.setRelationName(relation);
 
-	    PostgresSource ps = new PostgresSource(relation);
+	    PostgresSource ps = new PostgresSource(relation, relationPostgresSourceConfig);
 	    tasks.add(ps);
 	}
 	return tasks;
@@ -124,8 +124,12 @@ public String getRelationName() {
 
     @Override
     public List<Attribute> getAttributes() throws IOException, SQLException {
-	if (tableInfo.getTableAttributes() != null)
-	    return tableInfo.getTableAttributes();
+	if (!initialized) {
+	    initializeConnection();
+	}
+	if (this.attributes != null) {
+	    return this.attributes;
+	}
 	DatabaseMetaData metadata = connection.getMetaData();
 	ResultSet resultSet = metadata.getColumns(null, null, config.getRelationName(), null);
 	Vector<Attribute> attrs = new Vector<Attribute>();
@@ -137,7 +141,7 @@ public List<Attribute> getAttributes() throws IOException, SQLException {
 	    attrs.addElement(attr);
 	}
 	resultSet.close();
-	tableInfo.setTableAttributes(attrs);
+	this.attributes = attrs;
 	return attrs;
     }
 
@@ -163,6 +167,7 @@ private void initializeConnection() {
 
 	if (this.connectionPools.containsKey(connIdentifier)) {
 	    this.connection = this.connectionPools.get(connIdentifier);
+	    return;
 	}
 
 	try {
@@ -246,7 +251,7 @@ private boolean read(int num, List<Record> rec_list) throws SQLException {
 	    num--;
 	    // FIXME: profile and optimize this
 	    Record rec = new Record();
-	    for (int i = 0; i < this.tableInfo.getTableAttributes().size(); i++) {
+	    for (int i = 0; i < this.attributes.size(); i++) {
 		Object obj = theRS.getObject(i + 1);
 		if (obj != null) {
 		    String v1 = obj.toString();