cassandra_test.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690
  1. // Copyright (c) 2012 The gocql Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style
  3. // license that can be found in the LICENSE file.
  4. package gocql
  5. import (
  6. "bytes"
  7. "flag"
  8. "reflect"
  9. "speter.net/go/exp/math/dec/inf"
  10. "strconv"
  11. "strings"
  12. "sync"
  13. "testing"
  14. "time"
  15. )
  16. var (
  17. flagCluster = flag.String("cluster", "127.0.0.1", "a comma-separated list of host:port tuples")
  18. flagProto = flag.Int("proto", 2, "protcol version")
  19. flagCQL = flag.String("cql", "3.0.0", "CQL version")
  20. )
  21. var initOnce sync.Once
  22. func createSession(tb testing.TB) *Session {
  23. cluster := NewCluster(strings.Split(*flagCluster, ",")...)
  24. cluster.ProtoVersion = *flagProto
  25. cluster.CQLVersion = *flagCQL
  26. cluster.Authenticator = PasswordAuthenticator{
  27. Username: "cassandra",
  28. Password: "cassandra",
  29. }
  30. initOnce.Do(func() {
  31. session, err := cluster.CreateSession()
  32. if err != nil {
  33. tb.Fatal("createSession:", err)
  34. }
  35. // Drop and re-create the keyspace once. Different tests should use their own
  36. // individual tables, but can assume that the table does not exist before.
  37. if err := session.Query(`DROP KEYSPACE gocql_test`).Exec(); err != nil {
  38. tb.Log("drop keyspace:", err)
  39. }
  40. if err := session.Query(`CREATE KEYSPACE gocql_test
  41. WITH replication = {
  42. 'class' : 'SimpleStrategy',
  43. 'replication_factor' : 1
  44. }`).Exec(); err != nil {
  45. tb.Fatal("create keyspace:", err)
  46. }
  47. session.Close()
  48. })
  49. cluster.Keyspace = "gocql_test"
  50. session, err := cluster.CreateSession()
  51. if err != nil {
  52. tb.Fatal("createSession:", err)
  53. }
  54. return session
  55. }
  56. func TestEmptyHosts(t *testing.T) {
  57. cluster := NewCluster()
  58. if session, err := cluster.CreateSession(); err == nil {
  59. session.Close()
  60. t.Error("expected err, got nil")
  61. }
  62. }
  63. //TestUseStatementError checks to make sure the correct error is returned when the user tries to execute a use statement.
  64. func TestUseStatementError(t *testing.T) {
  65. session := createSession(t)
  66. defer session.Close()
  67. if err := session.Query("USE gocql_test").Exec(); err != nil {
  68. if err != ErrUseStmt {
  69. t.Error("expected ErrUseStmt, got " + err.Error())
  70. }
  71. } else {
  72. t.Error("expected err, got nil.")
  73. }
  74. }
  75. //TestInvalidKeyspace checks that an invalid keyspace will return promptly and without a flood of connections
  76. func TestInvalidKeyspace(t *testing.T) {
  77. cluster := NewCluster(strings.Split(*flagCluster, ",")...)
  78. cluster.ProtoVersion = *flagProto
  79. cluster.CQLVersion = *flagCQL
  80. cluster.Keyspace = "invalidKeyspace"
  81. session, err := cluster.CreateSession()
  82. if err != nil {
  83. if err != ErrNoConnectionsStarted {
  84. t.Errorf("Expected ErrNoConnections but got %v", err)
  85. }
  86. } else {
  87. session.Close() //Clean up the session
  88. t.Error("expected err, got nil.")
  89. }
  90. }
  91. func TestTracing(t *testing.T) {
  92. session := createSession(t)
  93. defer session.Close()
  94. if err := session.Query(`CREATE TABLE trace (id int primary key)`).Exec(); err != nil {
  95. t.Fatal("create:", err)
  96. }
  97. buf := &bytes.Buffer{}
  98. trace := NewTraceWriter(session, buf)
  99. if err := session.Query(`INSERT INTO trace (id) VALUES (?)`, 42).Trace(trace).Exec(); err != nil {
  100. t.Error("insert:", err)
  101. } else if buf.Len() == 0 {
  102. t.Error("insert: failed to obtain any tracing")
  103. }
  104. buf.Reset()
  105. var value int
  106. if err := session.Query(`SELECT id FROM trace WHERE id = ?`, 42).Trace(trace).Scan(&value); err != nil {
  107. t.Error("select:", err)
  108. } else if value != 42 {
  109. t.Errorf("value: expected %d, got %d", 42, value)
  110. } else if buf.Len() == 0 {
  111. t.Error("select: failed to obtain any tracing")
  112. }
  113. }
  114. func TestPaging(t *testing.T) {
  115. if *flagProto == 1 {
  116. t.Skip("Paging not supported. Please use Cassandra >= 2.0")
  117. }
  118. session := createSession(t)
  119. defer session.Close()
  120. if err := session.Query("CREATE TABLE paging (id int primary key)").Exec(); err != nil {
  121. t.Fatal("create table:", err)
  122. }
  123. for i := 0; i < 100; i++ {
  124. if err := session.Query("INSERT INTO paging (id) VALUES (?)", i).Exec(); err != nil {
  125. t.Fatal("insert:", err)
  126. }
  127. }
  128. iter := session.Query("SELECT id FROM paging").PageSize(10).Iter()
  129. var id int
  130. count := 0
  131. for iter.Scan(&id) {
  132. count++
  133. }
  134. if err := iter.Close(); err != nil {
  135. t.Fatal("close:", err)
  136. }
  137. if count != 100 {
  138. t.Fatalf("expected %d, got %d", 100, count)
  139. }
  140. }
  141. func TestCAS(t *testing.T) {
  142. if *flagProto == 1 {
  143. t.Skip("lightweight transactions not supported. Please use Cassandra >= 2.0")
  144. }
  145. session := createSession(t)
  146. defer session.Close()
  147. if err := session.Query(`CREATE TABLE cas_table (
  148. title varchar,
  149. revid timeuuid,
  150. PRIMARY KEY (title, revid)
  151. )`).Exec(); err != nil {
  152. t.Fatal("create:", err)
  153. }
  154. title, revid := "baz", TimeUUID()
  155. var titleCAS string
  156. var revidCAS UUID
  157. if applied, err := session.Query(`INSERT INTO cas_table (title, revid)
  158. VALUES (?, ?) IF NOT EXISTS`,
  159. title, revid).ScanCAS(&titleCAS, &revidCAS); err != nil {
  160. t.Fatal("insert:", err)
  161. } else if !applied {
  162. t.Fatal("insert should have been applied")
  163. }
  164. if applied, err := session.Query(`INSERT INTO cas_table (title, revid)
  165. VALUES (?, ?) IF NOT EXISTS`,
  166. title, revid).ScanCAS(&titleCAS, &revidCAS); err != nil {
  167. t.Fatal("insert:", err)
  168. } else if applied {
  169. t.Fatal("insert should not have been applied")
  170. } else if title != titleCAS || revid != revidCAS {
  171. t.Fatalf("expected %s/%v but got %s/%v", title, revid, titleCAS, revidCAS)
  172. }
  173. }
  174. func TestBatch(t *testing.T) {
  175. if *flagProto == 1 {
  176. t.Skip("atomic batches not supported. Please use Cassandra >= 2.0")
  177. }
  178. session := createSession(t)
  179. defer session.Close()
  180. if err := session.Query(`CREATE TABLE batch_table (id int primary key)`).Exec(); err != nil {
  181. t.Fatal("create table:", err)
  182. }
  183. batch := NewBatch(LoggedBatch)
  184. for i := 0; i < 100; i++ {
  185. batch.Query(`INSERT INTO batch_table (id) VALUES (?)`, i)
  186. }
  187. if err := session.ExecuteBatch(batch); err != nil {
  188. t.Fatal("execute batch:", err)
  189. }
  190. count := 0
  191. if err := session.Query(`SELECT COUNT(*) FROM batch_table`).Scan(&count); err != nil {
  192. t.Fatal("select count:", err)
  193. } else if count != 100 {
  194. t.Fatalf("count: expected %d, got %d\n", 100, count)
  195. }
  196. }
  197. // TestBatchLimit tests gocql to make sure batch operations larger than the maximum
  198. // statement limit are not submitted to a cassandra node.
  199. func TestBatchLimit(t *testing.T) {
  200. if *flagProto == 1 {
  201. t.Skip("atomic batches not supported. Please use Cassandra >= 2.0")
  202. }
  203. session := createSession(t)
  204. defer session.Close()
  205. if err := session.Query(`CREATE TABLE batch_table2 (id int primary key)`).Exec(); err != nil {
  206. t.Fatal("create table:", err)
  207. }
  208. batch := NewBatch(LoggedBatch)
  209. for i := 0; i < 65537; i++ {
  210. batch.Query(`INSERT INTO batch_table2 (id) VALUES (?)`, i)
  211. }
  212. if err := session.ExecuteBatch(batch); err != ErrTooManyStmts {
  213. t.Fatal("gocql attempted to execute a batch larger than the support limit of statements.")
  214. }
  215. }
  216. // TestTooManyQueryArgs tests to make sure the library correctly handles the application level bug
  217. // whereby too many query arguments are passed to a query
  218. func TestTooManyQueryArgs(t *testing.T) {
  219. if *flagProto == 1 {
  220. t.Skip("atomic batches not supported. Please use Cassandra >= 2.0")
  221. }
  222. session := createSession(t)
  223. defer session.Close()
  224. if err := session.Query(`CREATE TABLE too_many_query_args (id int primary key, value int)`).Exec(); err != nil {
  225. t.Fatal("create table:", err)
  226. }
  227. _, err := session.Query(`SELECT * FROM too_many_query_args WHERE id = ?`, 1, 2).Iter().SliceMap()
  228. if err == nil {
  229. t.Fatal("'`SELECT * FROM too_many_query_args WHERE id = ?`, 1, 2' should return an ErrQueryArgLength")
  230. }
  231. if err != ErrQueryArgLength {
  232. t.Fatalf("'`SELECT * FROM too_many_query_args WHERE id = ?`, 1, 2' should return an ErrQueryArgLength, but returned: %s", err)
  233. }
  234. batch := session.NewBatch(UnloggedBatch)
  235. batch.Query("INSERT INTO too_many_query_args (id, value) VALUES (?, ?)", 1, 2, 3)
  236. err = session.ExecuteBatch(batch)
  237. if err == nil {
  238. t.Fatal("'`INSERT INTO too_many_query_args (id, value) VALUES (?, ?)`, 1, 2, 3' should return an ErrQueryArgLength")
  239. }
  240. if err != ErrQueryArgLength {
  241. t.Fatalf("'INSERT INTO too_many_query_args (id, value) VALUES (?, ?)`, 1, 2, 3' should return an ErrQueryArgLength, but returned: %s", err)
  242. }
  243. }
  244. // TestNotEnoughQueryArgs tests to make sure the library correctly handles the application level bug
  245. // whereby not enough query arguments are passed to a query
  246. func TestNotEnoughQueryArgs(t *testing.T) {
  247. if *flagProto == 1 {
  248. t.Skip("atomic batches not supported. Please use Cassandra >= 2.0")
  249. }
  250. session := createSession(t)
  251. defer session.Close()
  252. if err := session.Query(`CREATE TABLE not_enough_query_args (id int, cluster int, value int, primary key (id, cluster))`).Exec(); err != nil {
  253. t.Fatal("create table:", err)
  254. }
  255. _, err := session.Query(`SELECT * FROM not_enough_query_args WHERE id = ? and cluster = ?`, 1).Iter().SliceMap()
  256. if err == nil {
  257. t.Fatal("'`SELECT * FROM not_enough_query_args WHERE id = ? and cluster = ?`, 1' should return an ErrQueryArgLength")
  258. }
  259. if err != ErrQueryArgLength {
  260. t.Fatalf("'`SELECT * FROM too_few_query_args WHERE id = ? and cluster = ?`, 1' should return an ErrQueryArgLength, but returned: %s", err)
  261. }
  262. batch := session.NewBatch(UnloggedBatch)
  263. batch.Query("INSERT INTO not_enough_query_args (id, cluster, value) VALUES (?, ?, ?)", 1, 2)
  264. err = session.ExecuteBatch(batch)
  265. if err == nil {
  266. t.Fatal("'`INSERT INTO not_enough_query_args (id, cluster, value) VALUES (?, ?, ?)`, 1, 2' should return an ErrQueryArgLength")
  267. }
  268. if err != ErrQueryArgLength {
  269. t.Fatalf("'INSERT INTO not_enough_query_args (id, cluster, value) VALUES (?, ?, ?)`, 1, 2' should return an ErrQueryArgLength, but returned: %s", err)
  270. }
  271. }
  272. // TestCreateSessionTimeout tests to make sure the CreateSession function timeouts out correctly
  273. // and prevents an infinite loop of connection retries.
  274. func TestCreateSessionTimeout(t *testing.T) {
  275. go func() {
  276. <-time.After(2 * time.Second)
  277. t.Fatal("no startup timeout")
  278. }()
  279. c := NewCluster("127.0.0.1:1")
  280. _, err := c.CreateSession()
  281. if err == nil {
  282. t.Fatal("expected ErrNoConnectionsStarted, but no error was returned.")
  283. }
  284. if err != ErrNoConnectionsStarted {
  285. t.Fatalf("expected ErrNoConnectionsStarted, but received %v", err)
  286. }
  287. }
  288. func TestSliceMap(t *testing.T) {
  289. session := createSession(t)
  290. defer session.Close()
  291. if err := session.Query(`CREATE TABLE slice_map_table (
  292. testuuid timeuuid PRIMARY KEY,
  293. testtimestamp timestamp,
  294. testvarchar varchar,
  295. testbigint bigint,
  296. testblob blob,
  297. testbool boolean,
  298. testfloat float,
  299. testdouble double,
  300. testint int,
  301. testdecimal decimal,
  302. testset set<int>,
  303. testmap map<varchar, varchar>
  304. )`).Exec(); err != nil {
  305. t.Fatal("create table:", err)
  306. }
  307. m := make(map[string]interface{})
  308. m["testuuid"] = TimeUUID()
  309. m["testvarchar"] = "Test VarChar"
  310. m["testbigint"] = time.Now().Unix()
  311. m["testtimestamp"] = time.Now().Truncate(time.Millisecond).UTC()
  312. m["testblob"] = []byte("test blob")
  313. m["testbool"] = true
  314. m["testfloat"] = float32(4.564)
  315. m["testdouble"] = float64(4.815162342)
  316. m["testint"] = 2343
  317. m["testdecimal"] = inf.NewDec(100, 0)
  318. m["testset"] = []int{1, 2, 3, 4, 5, 6, 7, 8, 9}
  319. m["testmap"] = map[string]string{"field1": "val1", "field2": "val2", "field3": "val3"}
  320. sliceMap := []map[string]interface{}{m}
  321. if err := session.Query(`INSERT INTO slice_map_table (testuuid, testtimestamp, testvarchar, testbigint, testblob, testbool, testfloat, testdouble, testint, testdecimal, testset, testmap) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
  322. m["testuuid"], m["testtimestamp"], m["testvarchar"], m["testbigint"], m["testblob"], m["testbool"], m["testfloat"], m["testdouble"], m["testint"], m["testdecimal"], m["testset"], m["testmap"]).Exec(); err != nil {
  323. t.Fatal("insert:", err)
  324. }
  325. if returned, retErr := session.Query(`SELECT * FROM slice_map_table`).Iter().SliceMap(); retErr != nil {
  326. t.Fatal("select:", retErr)
  327. } else {
  328. if sliceMap[0]["testuuid"] != returned[0]["testuuid"] {
  329. t.Fatal("returned testuuid did not match")
  330. }
  331. if sliceMap[0]["testtimestamp"] != returned[0]["testtimestamp"] {
  332. t.Fatalf("returned testtimestamp did not match: %v %v", sliceMap[0]["testtimestamp"], returned[0]["testtimestamp"])
  333. }
  334. if sliceMap[0]["testvarchar"] != returned[0]["testvarchar"] {
  335. t.Fatal("returned testvarchar did not match")
  336. }
  337. if sliceMap[0]["testbigint"] != returned[0]["testbigint"] {
  338. t.Fatal("returned testbigint did not match")
  339. }
  340. if !reflect.DeepEqual(sliceMap[0]["testblob"], returned[0]["testblob"]) {
  341. t.Fatal("returned testblob did not match")
  342. }
  343. if sliceMap[0]["testbool"] != returned[0]["testbool"] {
  344. t.Fatal("returned testbool did not match")
  345. }
  346. if sliceMap[0]["testfloat"] != returned[0]["testfloat"] {
  347. t.Fatal("returned testfloat did not match")
  348. }
  349. if sliceMap[0]["testdouble"] != returned[0]["testdouble"] {
  350. t.Fatal("returned testdouble did not match")
  351. }
  352. if sliceMap[0]["testint"] != returned[0]["testint"] {
  353. t.Fatal("returned testint did not match")
  354. }
  355. expectedDecimal := sliceMap[0]["testdecimal"].(*inf.Dec)
  356. returnedDecimal := returned[0]["testdecimal"].(*inf.Dec)
  357. if expectedDecimal.Cmp(returnedDecimal) != 0 {
  358. t.Fatal("returned testdecimal did not match")
  359. }
  360. if !reflect.DeepEqual(sliceMap[0]["testset"], returned[0]["testset"]) {
  361. t.Fatal("returned testset did not match")
  362. }
  363. if !reflect.DeepEqual(sliceMap[0]["testmap"], returned[0]["testmap"]) {
  364. t.Fatal("returned testmap did not match")
  365. }
  366. }
  367. // Test for MapScan()
  368. testMap := make(map[string]interface{})
  369. if !session.Query(`SELECT * FROM slice_map_table`).Iter().MapScan(testMap) {
  370. t.Fatal("MapScan failed to work with one row")
  371. }
  372. if sliceMap[0]["testuuid"] != testMap["testuuid"] {
  373. t.Fatal("returned testuuid did not match")
  374. }
  375. if sliceMap[0]["testtimestamp"] != testMap["testtimestamp"] {
  376. t.Fatal("returned testtimestamp did not match")
  377. }
  378. if sliceMap[0]["testvarchar"] != testMap["testvarchar"] {
  379. t.Fatal("returned testvarchar did not match")
  380. }
  381. if sliceMap[0]["testbigint"] != testMap["testbigint"] {
  382. t.Fatal("returned testbigint did not match")
  383. }
  384. if !reflect.DeepEqual(sliceMap[0]["testblob"], testMap["testblob"]) {
  385. t.Fatal("returned testblob did not match")
  386. }
  387. if sliceMap[0]["testbool"] != testMap["testbool"] {
  388. t.Fatal("returned testbool did not match")
  389. }
  390. if sliceMap[0]["testfloat"] != testMap["testfloat"] {
  391. t.Fatal("returned testfloat did not match")
  392. }
  393. if sliceMap[0]["testdouble"] != testMap["testdouble"] {
  394. t.Fatal("returned testdouble did not match")
  395. }
  396. if sliceMap[0]["testint"] != testMap["testint"] {
  397. t.Fatal("returned testint did not match")
  398. }
  399. expectedDecimal := sliceMap[0]["testdecimal"].(*inf.Dec)
  400. returnedDecimal := testMap["testdecimal"].(*inf.Dec)
  401. if expectedDecimal.Cmp(returnedDecimal) != 0 {
  402. t.Fatal("returned testdecimal did not match")
  403. }
  404. if !reflect.DeepEqual(sliceMap[0]["testset"], testMap["testset"]) {
  405. t.Fatal("returned testset did not match")
  406. }
  407. if !reflect.DeepEqual(sliceMap[0]["testmap"], testMap["testmap"]) {
  408. t.Fatal("returned testmap did not match")
  409. }
  410. }
  411. func TestScanWithNilArguments(t *testing.T) {
  412. session := createSession(t)
  413. defer session.Close()
  414. if err := session.Query(`CREATE TABLE scan_with_nil_arguments (
  415. foo varchar,
  416. bar int,
  417. PRIMARY KEY (foo, bar)
  418. )`).Exec(); err != nil {
  419. t.Fatal("create:", err)
  420. }
  421. for i := 1; i <= 20; i++ {
  422. if err := session.Query("INSERT INTO scan_with_nil_arguments (foo, bar) VALUES (?, ?)",
  423. "squares", i*i).Exec(); err != nil {
  424. t.Fatal("insert:", err)
  425. }
  426. }
  427. iter := session.Query("SELECT * FROM scan_with_nil_arguments WHERE foo = ?", "squares").Iter()
  428. var n int
  429. count := 0
  430. for iter.Scan(nil, &n) {
  431. count += n
  432. }
  433. if err := iter.Close(); err != nil {
  434. t.Fatal("close:", err)
  435. }
  436. if count != 2870 {
  437. t.Fatalf("expected %d, got %d", 2870, count)
  438. }
  439. }
  440. func TestScanCASWithNilArguments(t *testing.T) {
  441. if *flagProto == 1 {
  442. t.Skip("lightweight transactions not supported. Please use Cassandra >= 2.0")
  443. }
  444. session := createSession(t)
  445. defer session.Close()
  446. if err := session.Query(`CREATE TABLE scan_cas_with_nil_arguments (
  447. foo varchar,
  448. bar varchar,
  449. PRIMARY KEY (foo, bar)
  450. )`).Exec(); err != nil {
  451. t.Fatal("create:", err)
  452. }
  453. foo := "baz"
  454. var cas string
  455. if applied, err := session.Query(`INSERT INTO scan_cas_with_nil_arguments (foo, bar)
  456. VALUES (?, ?) IF NOT EXISTS`,
  457. foo, foo).ScanCAS(nil, nil); err != nil {
  458. t.Fatal("insert:", err)
  459. } else if !applied {
  460. t.Fatal("insert should have been applied")
  461. }
  462. if applied, err := session.Query(`INSERT INTO scan_cas_with_nil_arguments (foo, bar)
  463. VALUES (?, ?) IF NOT EXISTS`,
  464. foo, foo).ScanCAS(&cas, nil); err != nil {
  465. t.Fatal("insert:", err)
  466. } else if applied {
  467. t.Fatal("insert should not have been applied")
  468. } else if foo != cas {
  469. t.Fatalf("expected %v but got %v", foo, cas)
  470. }
  471. if applied, err := session.Query(`INSERT INTO scan_cas_with_nil_arguments (foo, bar)
  472. VALUES (?, ?) IF NOT EXISTS`,
  473. foo, foo).ScanCAS(nil, &cas); err != nil {
  474. t.Fatal("insert:", err)
  475. } else if applied {
  476. t.Fatal("insert should not have been applied")
  477. } else if foo != cas {
  478. t.Fatalf("expected %v but got %v", foo, cas)
  479. }
  480. }
  481. func injectInvalidPreparedStatement(t *testing.T, session *Session, table string) (string, *Conn) {
  482. if err := session.Query(`CREATE TABLE ` + table + ` (
  483. foo varchar,
  484. bar int,
  485. PRIMARY KEY (foo, bar)
  486. )`).Exec(); err != nil {
  487. t.Fatal("create:", err)
  488. }
  489. stmt := "INSERT INTO " + table + " (foo, bar) VALUES (?, 7)"
  490. conn := session.Pool.Pick(nil)
  491. flight := new(inflightPrepare)
  492. stmtsLRU.mu.Lock()
  493. stmtsLRU.lru.Add(conn.addr+stmt, flight)
  494. stmtsLRU.mu.Unlock()
  495. flight.info = &queryInfo{
  496. id: []byte{'f', 'o', 'o', 'b', 'a', 'r'},
  497. args: []ColumnInfo{ColumnInfo{
  498. Keyspace: "gocql_test",
  499. Table: table,
  500. Name: "foo",
  501. TypeInfo: &TypeInfo{
  502. Type: TypeVarchar,
  503. },
  504. }},
  505. }
  506. return stmt, conn
  507. }
  508. func TestReprepareStatement(t *testing.T) {
  509. session := createSession(t)
  510. defer session.Close()
  511. stmt, conn := injectInvalidPreparedStatement(t, session, "test_reprepare_statement")
  512. query := session.Query(stmt, "bar")
  513. if err := conn.executeQuery(query).Close(); err != nil {
  514. t.Fatalf("Failed to execute query for reprepare statement: %v", err)
  515. }
  516. }
  517. func TestReprepareBatch(t *testing.T) {
  518. if *flagProto == 1 {
  519. t.Skip("atomic batches not supported. Please use Cassandra >= 2.0")
  520. }
  521. session := createSession(t)
  522. defer session.Close()
  523. stmt, conn := injectInvalidPreparedStatement(t, session, "test_reprepare_statement_batch")
  524. batch := session.NewBatch(UnloggedBatch)
  525. batch.Query(stmt, "bar")
  526. if err := conn.executeBatch(batch); err != nil {
  527. t.Fatalf("Failed to execute query for reprepare statement: %v", err)
  528. }
  529. }
  530. //TestPreparedCacheEviction will make sure that the cache size is maintained
  531. func TestPreparedCacheEviction(t *testing.T) {
  532. session := createSession(t)
  533. defer session.Close()
  534. stmtsLRU.mu.Lock()
  535. stmtsLRU.Max(4)
  536. stmtsLRU.mu.Unlock()
  537. if err := session.Query("CREATE TABLE prepcachetest (id int,mod int,PRIMARY KEY (id))").Exec(); err != nil {
  538. t.Fatalf("failed to create table with error '%v'", err)
  539. }
  540. //Fill the table
  541. for i := 0; i < 2; i++ {
  542. if err := session.Query("INSERT INTO prepcachetest (id,mod) VALUES (?, ?)", i, 10000%(i+1)).Exec(); err != nil {
  543. t.Fatalf("insert into prepcachetest failed, err '%v'", err)
  544. }
  545. }
  546. //Populate the prepared statement cache with select statements
  547. var id, mod int
  548. for i := 0; i < 2; i++ {
  549. err := session.Query("SELECT id,mod FROM prepcachetest WHERE id = "+strconv.FormatInt(int64(i), 10)).Scan(&id, &mod)
  550. if err != nil {
  551. t.Fatalf("select from prepcachetest failed, error '%v'", err)
  552. }
  553. }
  554. //generate an update statement to test they are prepared
  555. err := session.Query("UPDATE prepcachetest SET mod = ? WHERE id = ?", 1, 11).Exec()
  556. if err != nil {
  557. t.Fatalf("update prepcachetest failed, error '%v'", err)
  558. }
  559. //generate a delete statement to test they are prepared
  560. err = session.Query("DELETE FROM prepcachetest WHERE id = ?", 1).Exec()
  561. if err != nil {
  562. t.Fatalf("delete from prepcachetest failed, error '%v'", err)
  563. }
  564. //generate an insert statement to test they are prepared
  565. err = session.Query("INSERT INTO prepcachetest (id,mod) VALUES (?, ?)", 3, 11).Exec()
  566. if err != nil {
  567. t.Fatalf("insert into prepcachetest failed, error '%v'", err)
  568. }
  569. //Make sure the cache size is maintained
  570. if stmtsLRU.lru.Len() != stmtsLRU.lru.MaxEntries {
  571. t.Fatalf("expected cache size of %v, got %v", stmtsLRU.lru.MaxEntries, stmtsLRU.lru.Len())
  572. }
  573. //Walk through all the configured hosts and test cache retention and eviction
  574. var selFound, insFound, updFound, delFound, selEvict bool
  575. for i := range session.cfg.Hosts {
  576. _, ok := stmtsLRU.lru.Get(session.cfg.Hosts[i] + ":9042SELECT id,mod FROM prepcachetest WHERE id = 1")
  577. selFound = selFound || ok
  578. _, ok = stmtsLRU.lru.Get(session.cfg.Hosts[i] + ":9042INSERT INTO prepcachetest (id,mod) VALUES (?, ?)")
  579. insFound = insFound || ok
  580. _, ok = stmtsLRU.lru.Get(session.cfg.Hosts[i] + ":9042UPDATE prepcachetest SET mod = ? WHERE id = ?")
  581. updFound = updFound || ok
  582. _, ok = stmtsLRU.lru.Get(session.cfg.Hosts[i] + ":9042DELETE FROM prepcachetest WHERE id = ?")
  583. delFound = delFound || ok
  584. _, ok = stmtsLRU.lru.Get(session.cfg.Hosts[i] + ":9042SELECT id,mod FROM prepcachetest WHERE id = 0")
  585. selEvict = selEvict || !ok
  586. }
  587. if !selEvict {
  588. t.Fatalf("expected first select statement to be purged, but statement was found in the cache.")
  589. }
  590. if !selFound {
  591. t.Fatalf("expected second select statement to be cached, but statement was purged or not prepared.")
  592. }
  593. if !insFound {
  594. t.Fatalf("expected insert statement to be cached, but statement was purged or not prepared.")
  595. }
  596. if !updFound {
  597. t.Fatalf("expected update statement to be cached, but statement was purged or not prepared.")
  598. }
  599. if !delFound {
  600. t.Error("expected delete statement to be cached, but statement was purged or not prepared.")
  601. }
  602. }