DatabaseDataSource


1   package prefuse.data.io.sql;
2   
3   import java.sql.Connection  ;
4   import java.sql.ResultSet  ;
5   import java.sql.ResultSetMetaData  ;
6   import java.sql.SQLException  ;
7   import java.sql.Statement  ;
8   import java.util.logging.Logger  ;
9   
10  import prefuse.data.Schema;
11  import prefuse.data.Table;
12  import prefuse.data.io.DataIOException;
13  import prefuse.data.util.Index;
14  
15  /**
16   * Sends queries to a relational database and processes the results, storing
17   * the results in prefuse Table instances. This class should not be
18   * instantiated directly. To access a database, the {@link ConnectionFactory}
19   * class should be used to retrieve an appropriate instance of this class.
20   * 
21   * @author <a HREF="http://jheer.org">jeffrey heer</a>
22   */
23  public class DatabaseDataSource {
24  
25      // logger
26      private static final Logger   s_logger 
27          = Logger.getLogger(DatabaseDataSource.class.getName());
28      
29      protected Connection         m_conn;
30      protected Statement          m_stmt;
31      protected SQLDataHandler m_handler;
32      
33      // ------------------------------------------------------------------------
34      
35      /**
36       * Creates a new DatabaseDataSource for reading data from a SQL relational
37       * database. This constructor is only package visible and is not intended
38       * for use by application level code. Instead, the
39       * {@link ConnectionFactory} class should be used to create any number of
40       * DatabaseDataSource connections.
41       */
42      DatabaseDataSource(Connection   conn, SQLDataHandler handler) {
43          m_conn = conn;
44          m_handler = handler;
45      }
46      
47      // ------------------------------------------------------------------------
48      // Synchronous Data Retrieval
49      
50      /**
51       * Executes a query and returns the results in a Table instance.
52       * @param query the text SQL query to execute
53       * @return a Table of the query results
54       * @throws DataIOException if an error occurs while executing the query 
55       * or adding the query results in a prefuse Table.
56       */
57      public synchronized Table getData(String   query) throws DataIOException {
58          return getData(null, query, null);
59      }
60  
61      /**
62       * Executes a query and returns the results in a Table instance.
63       * @param query the text SQL query to execute
64       * @param keyField the field to treat as a primary key, ensuring that this
65       *  field is indexed in the resulting table instance.
66       * @return a Table of the query results
67       * @throws DataIOException if an error occurs while executing the query 
68       * or adding the query results in a prefuse Table.
69       */
70      public synchronized Table getData(String   query, String   keyField)
71          throws DataIOException
72      {
73          return getData(null, query, keyField);
74      }
75      
76      /**
77       * Executes a query and returns the results in a Table instance.
78       * @param t the Table to store the results in. If this value is null, a
79       * new table will automatically be created.
80       * @param query the text SQL query to execute
81       * @return a Table of the query results
82       * @throws DataIOException if an error occurs while executing the query 
83       * or adding the query results in a prefuse Table.
84       */
85      public synchronized Table getData(Table t, String   query) 
86          throws DataIOException
87      {
88          return getData(t, query, null);
89      }
90      
91      /**
92       * Executes a query and returns the results in a Table instance.
93       * @param t the Table to store the results in. If this value is null, a
94       * new table will automatically be created.
95       * @param query the text SQL query to execute
96       * @param keyField used to determine if the row already exists in the table
97       * @return a Table of the query results
98       * @throws DataIOException if an error occurs while executing the query 
99       * or adding the query results in a prefuse Table.
100      */
101     public synchronized Table getData(Table t, String   query, String   keyField) 
102         throws DataIOException
103     {
104         return getData(t, query, keyField, null);
105     }
106     
107     /**
108      * Executes a query and returns the results in a Table instance.
109      * @param t the Table to store the results in. If this value is null, a
110      * new table will automatically be created.
111      * @param query the text SQL query to execute
112      * @param keyField used to determine if the row already exists in the table
113      * @param lock an optional Object to use as a lock when performing data
114      *  processing. This lock will be synchronized on whenever the Table is
115      *  modified.
116      * @return a Table of the query results
117      * @throws DataIOException if an error occurs while executing the query 
118      * or adding the query results in a prefuse Table.
119      */
120     public synchronized Table getData(Table t, String   query, 
121                                       String   keyField, Object   lock) 
122         throws DataIOException
123     {
124         ResultSet   rs;
125         try {
126             rs = executeQuery(query);
127         } catch ( SQLException   e ) {
128             throw new DataIOException(e);
129         }
130         return process(t, rs, keyField, lock);
131     }
132     
133     // ------------------------------------------------------------------------
134     // Asynchronous Data Retrieval
135 
136     /**
137      * Asynchronously executes a query and stores the results in the given 
138      * table instance. All data processing is done in a separate thread of
139      * execution.
140      * @param t the Table in which to store the results
141      * @param query the query to execute
142      */
143     public void loadData(Table t, String   query) {
144         loadData(t, query, null, null, null);
145     }
146 
147     /**
148      * Asynchronously executes a query and stores the results in the given 
149      * table instance. All data processing is done in a separate thread of
150      * execution.
151      * @param t the Table in which to store the results
152      * @param query the query to execute
153      * @param keyField the primary key field, comparisons on this field are
154      *  performed to recognize data records already present in the table.
155      */
156     public void loadData(Table t, String   query, String   keyField) {
157         loadData(t, query, keyField, null, null);
158     }
159     
160     /**
161      * Asynchronously executes a query and stores the results in the given 
162      * table instance. All data processing is done in a separate thread of
163      * execution.
164      * @param t the Table in which to store the results
165      * @param query the query to execute
166      * @param lock an optional Object to use as a lock when performing data
167      *  processing. This lock will be synchronized on whenever the Table is
168      *  modified.
169      */
170     public void loadData(Table t, String   query, Object   lock) {
171         loadData(t, query, null, lock, null);
172     }
173     
174     /**
175      * Asynchronously executes a query and stores the results in the given 
176      * table instance. All data processing is done in a separate thread of
177      * execution.
178      * @param t the Table in which to store the results
179      * @param query the query to execute
180      * @param keyField the primary key field, comparisons on this field are
181      *  performed to recognize data records already present in the table.
182      * @param lock an optional Object to use as a lock when performing data
183      *  processing. This lock will be synchronized on whenever the Table is
184      *  modified.
185      */
186     public void loadData(Table t, String   query, String   keyField, Object   lock) {
187         loadData(t, query, keyField, lock, null);
188     }
189     
190     /**
191      * Asynchronously executes a query and stores the results in the given 
192      * table instance. All data processing is done in a separate thread of
193      * execution.
194      * @param t the Table in which to store the results
195      * @param query the query to execute
196      * @param keyField the primary key field, comparisons on this field are
197      *  performed to recognize data records already present in the table.
198      *  A null value will result in no key checking.
199      * @param lock an optional Object to use as a lock when performing data
200      *  processing. This lock will be synchronized on whenever the Table is
201      *  modified. A null value will result in no locking.
202      * @param listener an optional listener that will provide notifications
203      *  before the query has been issued and after the query has been 
204      *  processed. This is most useful for post-processing operations.
205      */
206     public void loadData(Table t, String   query, String   keyField, 
207                          Object   lock, DataSourceWorker.Listener listener) {
208         DataSourceWorker.Entry e = new DataSourceWorker.Entry(
209                 this, t, query, keyField, lock, listener);
210         DataSourceWorker.submit(e);
211     }
212     
213     // ------------------------------------------------------------------------
214     
215     /**
216      * Execute a query and return the corresponding result set
217      * @param query the text SQL query to execute
218      * @return the ResultSet of the query
219      * @throws SQLException if an error occurs issuing the query
220      */
221     private ResultSet   executeQuery(String   query) throws SQLException   {
222         if ( m_stmt == null )
223             m_stmt = m_conn.createStatement();
224         
225         // clock in
226         long timein = System.currentTimeMillis();
227         
228         s_logger.info("Issuing query: "+query);
229         ResultSet   rset = m_stmt.executeQuery(query);
230         
231         // clock out
232         long time = System.currentTimeMillis()-timein;
233         s_logger.info("External query processing completed: "
234                 + (time/1000) + "." + (time%1000) + " seconds.");
235         
236         return rset;
237     }
238     
239     // ------------------------------------------------------------------------
240     
241     /**
242      * Process the results of a SQL query, putting retrieved data into a
243      * Table instance. If a null table is provided, a new table with the
244      * appropriate schema will be created.
245      * @param t the Table to store results in
246      * @param rset the SQL query result set
247      * @return a Table containing the query results
248      */
249     protected Table process(Table t, ResultSet   rset, String   key, Object   lock)
250         throws DataIOException
251     {
252         // clock in
253         int count = 0;
254         long timein = System.currentTimeMillis();
255 
256         try {
257             ResultSetMetaData   metadata = rset.getMetaData();
258             int ncols = metadata.getColumnCount();
259     
260             // create a new table if necessary
261             if ( t == null ) {
262                 t = getSchema(metadata, m_handler).instantiate();
263                 if ( key != null ) {
264                     try {
265                         t.index(key);
266                         s_logger.info("Indexed field: "+key);
267                     } catch ( Exception   e ) {
268                         s_logger.warning("Error indexing field: "+key);
269                     }
270                 }
271             }
272 
273             // set the lock, lock on the table itself if nothing else provided
274             lock = (lock==null ? t : lock);
275             
276             // process the returned rows
277             while ( rset.next() )
278             {
279                 synchronized ( lock ) {
280                     // determine the table row index to use
281                     int row = getExistingRow(t, rset, key);
282                     if ( row < 0 ) {
283                         row = t.addRow();
284                     }
285                     
286                     //process each value in the current row
287                     for ( int i=1; i<=ncols; ++i ) {
288                         m_handler.process(t, row, rset, i);
289                     }
290                 }
291                 
292                 // increment row count
293                 ++count;
294             }
295         } catch ( SQLException   e ) {
296             throw new DataIOException(e);
297         }
298         
299         // clock out
300         long time = System.currentTimeMillis()-timein;
301         s_logger.info("Internal query processing completed: "+count+" rows, "
302                 + (time/1000) + "." + (time%1000) + " seconds.");
303         
304         return t;
305     }
306     
307     /**
308      * See if a retrieved database row is already represented in the given
309      * Table.
310      * @param t the prefuse Table to check for an existing row
311      * @param rset the ResultSet, set to a particular row, which may or
312      * may not have a matching row in the prefuse Table
313      * @param keyField the key field to look up to check for an existing row
314      * @return the index of the existing row, or -1 if no match is found
315      * @throws SQLException
316      */
317     protected int getExistingRow(Table t, ResultSet   rset, String   keyField)
318         throws SQLException  
319     {
320         // check if we have a keyField, bail if not
321         if ( keyField == null )
322             return -1;
323         
324         // retrieve the column data type, bail if column is not found
325         Class   type = t.getColumnType(keyField);
326         if ( type == null )
327             return -1;
328         
329         // get the index and perform the lookup
330         Index index = t.index(keyField);
331         if ( type == int.class ) {
332             return index.get(rset.getInt(keyField));
333         } else if ( type == long.class ) {
334             return index.get(rset.getLong(keyField));
335         } else if ( type == float.class ) {
336             return index.get(rset.getFloat(keyField));
337         } else if ( type == double.class ) {
338             return index.get(rset.getDouble(keyField));
339         } else if ( !type.isPrimitive() ) {
340             return index.get(rset.getObject(keyField));
341         } else {
342             return -1;
343         }
344     }
345     
346     /**
347      * Given the metadata for a SQL result set and a data value handler for that
348      * result set, returns a corresponding schema for a prefuse table.
349      * @param metadata the SQL result set metadata
350      * @param handler the data value handler
351      * @return the schema determined by the metadata and handler
352      * @throws SQLException if an error occurs accessing the metadata
353      */
354     public Schema getSchema(ResultSetMetaData   metadata, SQLDataHandler handler)
355         throws SQLException  
356     {
357         int ncols = metadata.getColumnCount();
358         Schema schema = new Schema(ncols);
359         
360         // determine the table schema
361         for ( int i=1; i<=ncols; ++i ) {
362             String   name = metadata.getColumnName(i);
363             int sqlType = metadata.getColumnType(i);
364             Class   type = handler.getDataType(name, sqlType);
365             if ( type != null )
366                 schema.addColumn(name, type);
367         }
368         
369         return schema;
370     }
371 
372 } // end of class DatabaseDataSource
373
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags