OmniSciDB  a5dc49c757
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
Public Member Functions | Static Public Member Functions | Static Package Attributes | Private Member Functions | List of all members
com.mapd.tests.ForeignTableRefreshConcurrencyTest Class Reference
+ Collaboration diagram for com.mapd.tests.ForeignTableRefreshConcurrencyTest:

Public Member Functions

void runSqlAsUser (String sql, HeavyDBTestClient user, String logPrefix) throws Exception
 
void testConcurrency () throws Exception
 

Static Public Member Functions

static String getTimeStamp (int sec_from_now)
 
static void main (String[] args) throws Exception
 

Static Package Attributes

static final Logger logger
 

Private Member Functions

Path getAbsolutePath (String path)
 
void runTest (String db, String userName, String userPassword, int num_foreign_manual_refresh_threads, int num_foreign_scheduled_refresh_threads, int num_table_join_threads, int num_runs) throws Exception
 
void createForeignTestTableScheduledRefresh (HeavyDBTestClient dba, String foreign_table_name) throws Exception
 
void createForeignTestTableManualRefresh (HeavyDBTestClient dba, String foreign_table_name) throws Exception
 
void createTestTable (HeavyDBTestClient dba, String table_name, Path copy_from_path) throws Exception
 
void createTestTables (HeavyDBTestClient dba, String table_name, Path copy_from_path) throws Exception
 

Detailed Description

Definition at line 30 of file ForeignTableRefreshConcurrencyTest.java.

Member Function Documentation

void com.mapd.tests.ForeignTableRefreshConcurrencyTest.createForeignTestTableManualRefresh ( HeavyDBTestClient  dba,
String  foreign_table_name 
) throws Exception
inlineprivate

Definition at line 207 of file ForeignTableRefreshConcurrencyTest.java.

Referenced by com.mapd.tests.ForeignTableRefreshConcurrencyTest.testConcurrency().

208  {
209  dba.runSql("CREATE FOREIGN TABLE " + foreign_table_name + " "
210  + "(b BOOLEAN, t TINYINT, s SMALLINT, i INTEGER, bi BIGINT, f FLOAT, "
211  + "dc DECIMAL(10, 5), tm TIME, tp TIMESTAMP, d DATE, txt TEXT, "
212  + "txt_2 TEXT ENCODING NONE) "
213  + "SERVER test_server WITH "
214  + "(file_path = 'scalar_types.csv', "
215  + "fragment_size = 2);");
216  }

+ Here is the caller graph for this function:

void com.mapd.tests.ForeignTableRefreshConcurrencyTest.createForeignTestTableScheduledRefresh ( HeavyDBTestClient  dba,
String  foreign_table_name 
) throws Exception
inlineprivate

Definition at line 193 of file ForeignTableRefreshConcurrencyTest.java.

References com.mapd.tests.ForeignTableRefreshConcurrencyTest.getTimeStamp().

Referenced by com.mapd.tests.ForeignTableRefreshConcurrencyTest.testConcurrency().

194  {
195  dba.runSql("CREATE FOREIGN TABLE " + foreign_table_name + " "
196  + "(b BOOLEAN, t TINYINT, s SMALLINT, i INTEGER, bi BIGINT, f FLOAT, "
197  + "dc DECIMAL(10, 5), tm TIME, tp TIMESTAMP, d DATE, txt TEXT, "
198  + "txt_2 TEXT ENCODING NONE) "
199  + "SERVER test_server WITH "
200  + "(file_path = 'scalar_types.csv', "
201  + "refresh_timing_type = 'scheduled', "
202  + "refresh_start_date_time = '" + getTimeStamp(1) + "', "
203  + "refresh_interval = '1S', "
204  + "fragment_size = 2);");
205  }

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void com.mapd.tests.ForeignTableRefreshConcurrencyTest.createTestTable ( HeavyDBTestClient  dba,
String  table_name,
Path  copy_from_path 
) throws Exception
inlineprivate

Definition at line 218 of file ForeignTableRefreshConcurrencyTest.java.

Referenced by com.mapd.tests.ForeignTableRefreshConcurrencyTest.createTestTables().

220  {
221  dba.runSql("CREATE TABLE " + table_name
222  + " (id INTEGER, str TEXT ENCODING DICT(32), x DOUBLE, y BIGINT) WITH (FRAGMENT_SIZE=1, partitions='replicated')");
223  dba.runSql("COPY " + table_name + " FROM '" + copy_from_path.toString()
224  + "' WITH (header='false');");
225  }

+ Here is the caller graph for this function:

void com.mapd.tests.ForeignTableRefreshConcurrencyTest.createTestTables ( HeavyDBTestClient  dba,
String  table_name,
Path  copy_from_path 
) throws Exception
inlineprivate

Definition at line 227 of file ForeignTableRefreshConcurrencyTest.java.

References com.mapd.tests.ForeignTableRefreshConcurrencyTest.createTestTable().

Referenced by com.mapd.tests.ForeignTableRefreshConcurrencyTest.testConcurrency().

229  {
230  createTestTable(dba, table_name + "_left", copy_from_path);
231  createTestTable(dba, table_name + "_right", copy_from_path);
232  }
com.mapd.tests.ForeignTableRefreshConcurrencyTest.createTestTable
void createTestTable(HeavyDBTestClient dba, String table_name, Path copy_from_path)
Definition: ForeignTableRefreshConcurrencyTest.java:218

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

Path com.mapd.tests.ForeignTableRefreshConcurrencyTest.getAbsolutePath ( String  path)
inlineprivate

Definition at line 48 of file ForeignTableRefreshConcurrencyTest.java.

Referenced by com.mapd.tests.ForeignTableRefreshConcurrencyTest.testConcurrency().

48  {
49  Path path_obj = Paths.get(path).toAbsolutePath();
50  assert Files.exists(path_obj);
51  return path_obj;
52  }

+ Here is the caller graph for this function:

static String com.mapd.tests.ForeignTableRefreshConcurrencyTest.getTimeStamp ( int  sec_from_now)
inlinestatic

Definition at line 34 of file ForeignTableRefreshConcurrencyTest.java.

References heavydb.dtypes.Date.

Referenced by com.mapd.tests.ForeignTableRefreshConcurrencyTest.createForeignTestTableScheduledRefresh().

34  {
35  Calendar cal = Calendar.getInstance();
36  cal.setTime(new Date());
37  cal.add(Calendar.SECOND, sec_from_now);
38  SimpleDateFormat date_format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
39  date_format.setTimeZone(TimeZone.getTimeZone("GMT"));
40  return date_format.format(cal.getTime());
41  }

+ Here is the caller graph for this function:

static void com.mapd.tests.ForeignTableRefreshConcurrencyTest.main ( String[]  args) throws Exception
inlinestatic

Definition at line 43 of file ForeignTableRefreshConcurrencyTest.java.

43  {
44  ForeignTableRefreshConcurrencyTest test = new ForeignTableRefreshConcurrencyTest();
45  test.testConcurrency();
46  }
void com.mapd.tests.ForeignTableRefreshConcurrencyTest.runSqlAsUser ( String  sql,
HeavyDBTestClient  user,
String  logPrefix 
) throws Exception
inline

Definition at line 187 of file ForeignTableRefreshConcurrencyTest.java.

Referenced by com.mapd.tests.ForeignTableRefreshConcurrencyTest.runTest().

188  {
189  logger.info(logPrefix + " " + sql);
190  user.runSql(sql);
191  }

+ Here is the caller graph for this function:

void com.mapd.tests.ForeignTableRefreshConcurrencyTest.runTest ( String  db,
String  userName,
String  userPassword,
int  num_foreign_manual_refresh_threads,
int  num_foreign_scheduled_refresh_threads,
int  num_table_join_threads,
int  num_runs 
) throws Exception
inlineprivate

Definition at line 60 of file ForeignTableRefreshConcurrencyTest.java.

References run, and com.mapd.tests.ForeignTableRefreshConcurrencyTest.runSqlAsUser().

Referenced by com.mapd.tests.ForeignTableRefreshConcurrencyTest.testConcurrency().

66  {
67  ArrayList<Exception> exceptions = new ArrayList<Exception>();
68 
69  Thread[] foreign_table_manual_refresh_threads =
70  new Thread[num_foreign_manual_refresh_threads];
71  Thread[] foreign_table_scheduled_refresh_threads =
72  new Thread[num_foreign_scheduled_refresh_threads];
73  Thread[] table_join_threads = new Thread[num_table_join_threads];
74 
75  for (int i = 0; i < num_foreign_manual_refresh_threads; ++i) {
76  final int tid = i;
77  foreign_table_manual_refresh_threads[tid] = new Thread(new Runnable() {
78  @Override
79  public void run() {
80  final String thread_name = "F[" + tid + "]";
81  try {
82  logger.info("Starting manual foreign table refresh thread " + thread_name);
83  HeavyDBTestClient user = HeavyDBTestClient.getClient(
84  "localhost", 6274, db, userName, userPassword);
85  for (int irun = 0; irun < num_runs; ++irun) {
86  runSqlAsUser("SELECT * FROM test_foreign_table_manual_refresh_" + tid + ";",
87  user,
88  thread_name);
89  runSqlAsUser("REFRESH FOREIGN TABLES test_foreign_table_manual_refresh_"
90  + tid + ";",
91  user,
92  thread_name);
93  }
94  logger.info("Finished foreign table manual refresh " + thread_name);
95  } catch (Exception e) {
96  logger.error("Foreign table refresh " + thread_name
97  + " Caught Exception: " + e.getMessage(),
98  e);
99  exceptions.add(e);
100  }
101  }
102  });
103  foreign_table_manual_refresh_threads[tid].start();
104  }
105 
106  for (int i = 0; i < num_foreign_scheduled_refresh_threads; ++i) {
107  final int tid = i;
108  foreign_table_scheduled_refresh_threads[tid] = new Thread(new Runnable() {
109  @Override
110  public void run() {
111  final String thread_name = "S[" + tid + "]";
112  try {
113  logger.info("Starting scheduled foreign table refresh thread " + thread_name);
114  HeavyDBTestClient user = HeavyDBTestClient.getClient(
115  "localhost", 6274, db, userName, userPassword);
116  for (int irun = 0; irun < num_runs; ++irun) {
117  runSqlAsUser(
118  "SELECT * FROM test_foreign_table_scheduled_refresh_" + tid + ";",
119  user,
120  thread_name);
121  Thread.sleep(1000); // sleep 1s between queries to allow
122  // refreshes to occur, which occur at 1s intervals
123  }
124  logger.info("Finished foreign table scheduled refresh " + thread_name);
125  } catch (Exception e) {
126  logger.error("Foreign table scheduled refresh " + thread_name
127  + " Caught Exception: " + e.getMessage(),
128  e);
129  exceptions.add(e);
130  }
131  }
132  });
133  foreign_table_scheduled_refresh_threads[tid].start();
134  }
135 
136  for (int i = 0; i < num_table_join_threads; ++i) {
137  final int tid = i;
138  table_join_threads[tid] = new Thread(new Runnable() {
139  @Override
140  public void run() {
141  final String thread_name = "T[" + tid + "]";
142  try {
143  logger.info("Starting table join " + thread_name);
144  HeavyDBTestClient user = HeavyDBTestClient.getClient(
145  "localhost", 6274, db, userName, userPassword);
146  for (int irun = 0; irun < num_runs; ++irun) {
147  runSqlAsUser("SELECT * FROM test_table_" + tid
148  + "_left AS l JOIN test_table_" + tid
149  + "_right AS r ON l.id = r.id;",
150  user,
151  thread_name);
152  Thread.sleep(1000); // sleep 1s between queries to allow more
153  // contention between foreign table refreshes;
154  // especially in the case of scheduled refreshes
155  }
156  logger.info("Finished table join thread T[" + tid + "]");
157  } catch (Exception e) {
158  e.printStackTrace();
159  logger.error(
160  "Table join " + thread_name + " Caught Exception: " + e.getMessage(),
161  e);
162  exceptions.add(e);
163  }
164  }
165  });
166  table_join_threads[tid].start();
167  }
168 
169  for (Thread t : foreign_table_manual_refresh_threads) {
170  t.join();
171  }
172  for (Thread t : foreign_table_scheduled_refresh_threads) {
173  t.join();
174  }
175  for (Thread t : table_join_threads) {
176  t.join();
177  }
178 
179  for (Exception e : exceptions) {
180  if (null != e) {
181  logger.error("Exception: " + e.getMessage(), e);
182  throw e;
183  }
184  }
185  }
com.mapd.tests.ForeignTableRefreshConcurrencyTest.runSqlAsUser
void runSqlAsUser(String sql, HeavyDBTestClient user, String logPrefix)
Definition: ForeignTableRefreshConcurrencyTest.java:187
run
static bool run
Definition: KafkaImporter.cpp:49

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void com.mapd.tests.ForeignTableRefreshConcurrencyTest.testConcurrency ( ) throws Exception
inline

Definition at line 234 of file ForeignTableRefreshConcurrencyTest.java.

References com.mapd.tests.ForeignTableRefreshConcurrencyTest.createForeignTestTableManualRefresh(), com.mapd.tests.ForeignTableRefreshConcurrencyTest.createForeignTestTableScheduledRefresh(), com.mapd.tests.ForeignTableRefreshConcurrencyTest.createTestTables(), com.mapd.tests.ForeignTableRefreshConcurrencyTest.getAbsolutePath(), and com.mapd.tests.ForeignTableRefreshConcurrencyTest.runTest().

234  {
235  logger.info("ForeignTableRefreshConcurrencyTest()");
236 
237  HeavyDBTestClient su = HeavyDBTestClient.getClient(
238  "localhost", 6274, "heavyai", "admin", "HyperInteractive");
239 
240  // initialize
241  su.runSql("DROP DATABASE IF EXISTS db1;");
242  su.runSql("CREATE DATABASE db1;");
243 
244  // create tables for use
245  final int num_foreign_manual_refresh_threads = 2;
246  final int num_foreign_scheduled_refresh_threads = 2;
247  final int num_table_join_threads = 2;
248  Path table_import_path = getAbsolutePath(
249  "../java/utility/src/main/java/com/mapd/tests/data/simple_test.csv");
250  Path foreign_server_path = getAbsolutePath("../Tests/FsiDataFiles/");
251  HeavyDBTestClient dba = HeavyDBTestClient.getClient(
252  "localhost", 6274, "db1", "admin", "HyperInteractive");
253  dba.runSql("CREATE SERVER test_server "
254  + "FOREIGN DATA WRAPPER delimited_file WITH (storage_type = 'LOCAL_FILE', "
255  + "base_path = '" + foreign_server_path.toString() + "');");
256  for (int i = 0; i < num_foreign_manual_refresh_threads; ++i) {
257  createForeignTestTableManualRefresh(dba, "test_foreign_table_manual_refresh_" + i);
258  }
259  for (int i = 0; i < num_foreign_scheduled_refresh_threads; ++i) {
260  createForeignTestTableScheduledRefresh(
261  dba, "test_foreign_table_scheduled_refresh_" + i);
262  }
263  for (int i = 0; i < num_table_join_threads; ++i) {
264  createTestTables(dba, "test_table_" + i, table_import_path);
265  }
266 
267  runTest("db1",
268  "admin",
269  "HyperInteractive",
270  num_foreign_manual_refresh_threads,
271  num_foreign_scheduled_refresh_threads,
272  num_table_join_threads,
273  25);
274 
275  // cleanup
276  for (int i = 0; i < num_foreign_manual_refresh_threads; ++i) {
277  dba.runSql("DROP FOREIGN TABLE test_foreign_table_manual_refresh_" + i + ";");
278  }
279  for (int i = 0; i < num_foreign_manual_refresh_threads; ++i) {
280  dba.runSql("DROP FOREIGN TABLE test_foreign_table_scheduled_refresh_" + i + ";");
281  }
282  for (int i = 0; i < num_table_join_threads; ++i) {
283  dba.runSql("DROP TABLE test_table_" + i + "_left ;");
284  dba.runSql("DROP TABLE test_table_" + i + "_right ;");
285  }
286  dba.runSql("DROP SERVER test_server;");
287  su.runSql("DROP DATABASE db1;");
288 
289  logger.info("ForeignTableRefreshConcurrencyTest() done");
290  }
com.mapd.tests.ForeignTableRefreshConcurrencyTest.createForeignTestTableManualRefresh
void createForeignTestTableManualRefresh(HeavyDBTestClient dba, String foreign_table_name)
Definition: ForeignTableRefreshConcurrencyTest.java:207
com.mapd.tests.ForeignTableRefreshConcurrencyTest.createTestTables
void createTestTables(HeavyDBTestClient dba, String table_name, Path copy_from_path)
Definition: ForeignTableRefreshConcurrencyTest.java:227
com.mapd.tests.ForeignTableRefreshConcurrencyTest.runTest
void runTest(String db, String userName, String userPassword, int num_foreign_manual_refresh_threads, int num_foreign_scheduled_refresh_threads, int num_table_join_threads, int num_runs)
Definition: ForeignTableRefreshConcurrencyTest.java:60
com.mapd.tests.ForeignTableRefreshConcurrencyTest.createForeignTestTableScheduledRefresh
void createForeignTestTableScheduledRefresh(HeavyDBTestClient dba, String foreign_table_name)
Definition: ForeignTableRefreshConcurrencyTest.java:193

+ Here is the call graph for this function:

Member Data Documentation

final Logger com.mapd.tests.ForeignTableRefreshConcurrencyTest.logger
staticpackage
Initial value:
=
LoggerFactory.getLogger(ForeignTableRefreshConcurrencyTest.class)

Definition at line 31 of file ForeignTableRefreshConcurrencyTest.java.

The documentation for this class was generated from the following file: