metadata.go 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086
  1. // Copyright (c) 2015 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. "encoding/hex"
  7. "encoding/json"
  8. "fmt"
  9. "log"
  10. "strconv"
  11. "strings"
  12. "sync"
  13. )
  14. // schema metadata for a keyspace
  15. type KeyspaceMetadata struct {
  16. Name string
  17. DurableWrites bool
  18. StrategyClass string
  19. StrategyOptions map[string]interface{}
  20. Tables map[string]*TableMetadata
  21. }
  22. // schema metadata for a table (a.k.a. column family)
  23. type TableMetadata struct {
  24. Keyspace string
  25. Name string
  26. KeyValidator string
  27. Comparator string
  28. DefaultValidator string
  29. KeyAliases []string
  30. ColumnAliases []string
  31. ValueAlias string
  32. PartitionKey []*ColumnMetadata
  33. ClusteringColumns []*ColumnMetadata
  34. Columns map[string]*ColumnMetadata
  35. OrderedColumns []string
  36. }
  37. // schema metadata for a column
  38. type ColumnMetadata struct {
  39. Keyspace string
  40. Table string
  41. Name string
  42. ComponentIndex int
  43. Kind ColumnKind
  44. Validator string
  45. Type TypeInfo
  46. ClusteringOrder string
  47. Order ColumnOrder
  48. Index ColumnIndexMetadata
  49. }
  50. // the ordering of the column with regard to its comparator
  51. type ColumnOrder bool
  52. const (
  53. ASC ColumnOrder = false
  54. DESC = true
  55. )
  56. type ColumnIndexMetadata struct {
  57. Name string
  58. Type string
  59. Options map[string]interface{}
  60. }
  61. type ColumnKind int
  62. const (
  63. ColumnUnkownKind ColumnKind = iota
  64. ColumnPartitionKey
  65. ColumnClusteringKey
  66. ColumnRegular
  67. ColumnCompact
  68. ColumnStatic
  69. )
  70. func (c ColumnKind) String() string {
  71. switch c {
  72. case ColumnPartitionKey:
  73. return "partition_key"
  74. case ColumnClusteringKey:
  75. return "clustering_key"
  76. case ColumnRegular:
  77. return "regular"
  78. case ColumnCompact:
  79. return "compact"
  80. case ColumnStatic:
  81. return "static"
  82. default:
  83. return fmt.Sprintf("unkown_column_%d", c)
  84. }
  85. }
  86. func (c *ColumnKind) UnmarshalCQL(typ TypeInfo, p []byte) error {
  87. if typ.Type() != TypeVarchar {
  88. return unmarshalErrorf("unable to marshall %s into ColumnKind, expected Varchar", typ)
  89. }
  90. kind, err := columnKindFromSchema(string(p))
  91. if err != nil {
  92. return err
  93. }
  94. *c = kind
  95. return nil
  96. }
  97. func columnKindFromSchema(kind string) (ColumnKind, error) {
  98. switch kind {
  99. case "partition_key":
  100. return ColumnPartitionKey, nil
  101. case "clustering_key", "clustering":
  102. return ColumnClusteringKey, nil
  103. case "regular":
  104. return ColumnRegular, nil
  105. case "compact_value":
  106. return ColumnCompact, nil
  107. case "static":
  108. return ColumnStatic, nil
  109. default:
  110. return -1, fmt.Errorf("unknown column kind: %q", kind)
  111. }
  112. }
  113. // default alias values
  114. const (
  115. DEFAULT_KEY_ALIAS = "key"
  116. DEFAULT_COLUMN_ALIAS = "column"
  117. DEFAULT_VALUE_ALIAS = "value"
  118. )
  119. // queries the cluster for schema information for a specific keyspace
  120. type schemaDescriber struct {
  121. session *Session
  122. mu sync.Mutex
  123. cache map[string]*KeyspaceMetadata
  124. }
  125. // creates a session bound schema describer which will query and cache
  126. // keyspace metadata
  127. func newSchemaDescriber(session *Session) *schemaDescriber {
  128. return &schemaDescriber{
  129. session: session,
  130. cache: map[string]*KeyspaceMetadata{},
  131. }
  132. }
  133. // returns the cached KeyspaceMetadata held by the describer for the named
  134. // keyspace.
  135. func (s *schemaDescriber) getSchema(keyspaceName string) (*KeyspaceMetadata, error) {
  136. s.mu.Lock()
  137. defer s.mu.Unlock()
  138. metadata, found := s.cache[keyspaceName]
  139. if !found {
  140. // refresh the cache for this keyspace
  141. err := s.refreshSchema(keyspaceName)
  142. if err != nil {
  143. return nil, err
  144. }
  145. metadata = s.cache[keyspaceName]
  146. }
  147. return metadata, nil
  148. }
  149. // clears the already cached keyspace metadata
  150. func (s *schemaDescriber) clearSchema(keyspaceName string) {
  151. s.mu.Lock()
  152. defer s.mu.Unlock()
  153. delete(s.cache, keyspaceName)
  154. }
  155. // forcibly updates the current KeyspaceMetadata held by the schema describer
  156. // for a given named keyspace.
  157. func (s *schemaDescriber) refreshSchema(keyspaceName string) error {
  158. var err error
  159. // query the system keyspace for schema data
  160. // TODO retrieve concurrently
  161. keyspace, err := getKeyspaceMetadata(s.session, keyspaceName)
  162. if err != nil {
  163. return err
  164. }
  165. tables, err := getTableMetadata(s.session, keyspaceName)
  166. if err != nil {
  167. return err
  168. }
  169. columns, err := getColumnMetadata(s.session, keyspaceName)
  170. if err != nil {
  171. return err
  172. }
  173. // organize the schema data
  174. compileMetadata(s.session.cfg.ProtoVersion, keyspace, tables, columns)
  175. // update the cache
  176. s.cache[keyspaceName] = keyspace
  177. return nil
  178. }
  179. // "compiles" derived information about keyspace, table, and column metadata
  180. // for a keyspace from the basic queried metadata objects returned by
  181. // getKeyspaceMetadata, getTableMetadata, and getColumnMetadata respectively;
  182. // Links the metadata objects together and derives the column composition of
  183. // the partition key and clustering key for a table.
  184. func compileMetadata(
  185. protoVersion int,
  186. keyspace *KeyspaceMetadata,
  187. tables []TableMetadata,
  188. columns []ColumnMetadata,
  189. ) {
  190. keyspace.Tables = make(map[string]*TableMetadata)
  191. for i := range tables {
  192. tables[i].Columns = make(map[string]*ColumnMetadata)
  193. keyspace.Tables[tables[i].Name] = &tables[i]
  194. }
  195. // add columns from the schema data
  196. for i := range columns {
  197. // decode the validator for TypeInfo and order
  198. if columns[i].ClusteringOrder != "" { // Cassandra 3.x+
  199. columns[i].Type = NativeType{typ: getCassandraType(columns[i].Validator)}
  200. columns[i].Order = ASC
  201. if columns[i].ClusteringOrder == "desc" {
  202. columns[i].Order = DESC
  203. }
  204. } else {
  205. validatorParsed := parseType(columns[i].Validator)
  206. columns[i].Type = validatorParsed.types[0]
  207. columns[i].Order = ASC
  208. if validatorParsed.reversed[0] {
  209. columns[i].Order = DESC
  210. }
  211. }
  212. table := keyspace.Tables[columns[i].Table]
  213. table.Columns[columns[i].Name] = &columns[i]
  214. table.OrderedColumns = append(table.OrderedColumns, columns[i].Name)
  215. }
  216. if protoVersion == 1 {
  217. compileV1Metadata(tables)
  218. } else {
  219. compileV2Metadata(tables)
  220. }
  221. }
  222. // Compiles derived information from TableMetadata which have had
  223. // ColumnMetadata added already. V1 protocol does not return as much
  224. // column metadata as V2+ (because V1 doesn't support the "type" column in the
  225. // system.schema_columns table) so determining PartitionKey and ClusterColumns
  226. // is more complex.
  227. func compileV1Metadata(tables []TableMetadata) {
  228. for i := range tables {
  229. table := &tables[i]
  230. // decode the key validator
  231. keyValidatorParsed := parseType(table.KeyValidator)
  232. // decode the comparator
  233. comparatorParsed := parseType(table.Comparator)
  234. // the partition key length is the same as the number of types in the
  235. // key validator
  236. table.PartitionKey = make([]*ColumnMetadata, len(keyValidatorParsed.types))
  237. // V1 protocol only returns "regular" columns from
  238. // system.schema_columns (there is no type field for columns)
  239. // so the alias information is used to
  240. // create the partition key and clustering columns
  241. // construct the partition key from the alias
  242. for i := range table.PartitionKey {
  243. var alias string
  244. if len(table.KeyAliases) > i {
  245. alias = table.KeyAliases[i]
  246. } else if i == 0 {
  247. alias = DEFAULT_KEY_ALIAS
  248. } else {
  249. alias = DEFAULT_KEY_ALIAS + strconv.Itoa(i+1)
  250. }
  251. column := &ColumnMetadata{
  252. Keyspace: table.Keyspace,
  253. Table: table.Name,
  254. Name: alias,
  255. Type: keyValidatorParsed.types[i],
  256. Kind: ColumnPartitionKey,
  257. ComponentIndex: i,
  258. }
  259. table.PartitionKey[i] = column
  260. table.Columns[alias] = column
  261. }
  262. // determine the number of clustering columns
  263. size := len(comparatorParsed.types)
  264. if comparatorParsed.isComposite {
  265. if len(comparatorParsed.collections) != 0 ||
  266. (len(table.ColumnAliases) == size-1 &&
  267. comparatorParsed.types[size-1].Type() == TypeVarchar) {
  268. size = size - 1
  269. }
  270. } else {
  271. if !(len(table.ColumnAliases) != 0 || len(table.Columns) == 0) {
  272. size = 0
  273. }
  274. }
  275. table.ClusteringColumns = make([]*ColumnMetadata, size)
  276. for i := range table.ClusteringColumns {
  277. var alias string
  278. if len(table.ColumnAliases) > i {
  279. alias = table.ColumnAliases[i]
  280. } else if i == 0 {
  281. alias = DEFAULT_COLUMN_ALIAS
  282. } else {
  283. alias = DEFAULT_COLUMN_ALIAS + strconv.Itoa(i+1)
  284. }
  285. order := ASC
  286. if comparatorParsed.reversed[i] {
  287. order = DESC
  288. }
  289. column := &ColumnMetadata{
  290. Keyspace: table.Keyspace,
  291. Table: table.Name,
  292. Name: alias,
  293. Type: comparatorParsed.types[i],
  294. Order: order,
  295. Kind: ColumnClusteringKey,
  296. ComponentIndex: i,
  297. }
  298. table.ClusteringColumns[i] = column
  299. table.Columns[alias] = column
  300. }
  301. if size != len(comparatorParsed.types)-1 {
  302. alias := DEFAULT_VALUE_ALIAS
  303. if len(table.ValueAlias) > 0 {
  304. alias = table.ValueAlias
  305. }
  306. // decode the default validator
  307. defaultValidatorParsed := parseType(table.DefaultValidator)
  308. column := &ColumnMetadata{
  309. Keyspace: table.Keyspace,
  310. Table: table.Name,
  311. Name: alias,
  312. Type: defaultValidatorParsed.types[0],
  313. Kind: ColumnRegular,
  314. }
  315. table.Columns[alias] = column
  316. }
  317. }
  318. }
  319. // The simpler compile case for V2+ protocol
  320. func compileV2Metadata(tables []TableMetadata) {
  321. for i := range tables {
  322. table := &tables[i]
  323. clusteringColumnCount := componentColumnCountOfType(table.Columns, ColumnClusteringKey)
  324. table.ClusteringColumns = make([]*ColumnMetadata, clusteringColumnCount)
  325. if table.KeyValidator != "" {
  326. keyValidatorParsed := parseType(table.KeyValidator)
  327. table.PartitionKey = make([]*ColumnMetadata, len(keyValidatorParsed.types))
  328. } else { // Cassandra 3.x+
  329. partitionKeyCount := componentColumnCountOfType(table.Columns, ColumnPartitionKey)
  330. table.PartitionKey = make([]*ColumnMetadata, partitionKeyCount)
  331. }
  332. for _, columnName := range table.OrderedColumns {
  333. column := table.Columns[columnName]
  334. if column.Kind == ColumnPartitionKey {
  335. table.PartitionKey[column.ComponentIndex] = column
  336. } else if column.Kind == ColumnClusteringKey {
  337. table.ClusteringColumns[column.ComponentIndex] = column
  338. }
  339. }
  340. }
  341. }
  342. // returns the count of coluns with the given "kind" value.
  343. func componentColumnCountOfType(columns map[string]*ColumnMetadata, kind ColumnKind) int {
  344. maxComponentIndex := -1
  345. for _, column := range columns {
  346. if column.Kind == kind && column.ComponentIndex > maxComponentIndex {
  347. maxComponentIndex = column.ComponentIndex
  348. }
  349. }
  350. return maxComponentIndex + 1
  351. }
  352. // query only for the keyspace metadata for the specified keyspace from system.schema_keyspace
  353. func getKeyspaceMetadata(session *Session, keyspaceName string) (*KeyspaceMetadata, error) {
  354. keyspace := &KeyspaceMetadata{Name: keyspaceName}
  355. if session.useSystemSchema { // Cassandra 3.x+
  356. const stmt = `
  357. SELECT durable_writes, replication
  358. FROM system_schema.keyspaces
  359. WHERE keyspace_name = ?`
  360. var replication map[string]string
  361. iter := session.control.query(stmt, keyspaceName)
  362. iter.Scan(&keyspace.DurableWrites, &replication)
  363. err := iter.Close()
  364. if err != nil {
  365. return nil, fmt.Errorf("Error querying keyspace schema: %v", err)
  366. }
  367. keyspace.StrategyClass = replication["class"]
  368. keyspace.StrategyOptions = make(map[string]interface{})
  369. for k, v := range replication {
  370. keyspace.StrategyOptions[k] = v
  371. }
  372. } else {
  373. const stmt = `
  374. SELECT durable_writes, strategy_class, strategy_options
  375. FROM system.schema_keyspaces
  376. WHERE keyspace_name = ?`
  377. var strategyOptionsJSON []byte
  378. iter := session.control.query(stmt, keyspaceName)
  379. iter.Scan(&keyspace.DurableWrites, &keyspace.StrategyClass, &strategyOptionsJSON)
  380. err := iter.Close()
  381. if err != nil {
  382. return nil, fmt.Errorf("Error querying keyspace schema: %v", err)
  383. }
  384. err = json.Unmarshal(strategyOptionsJSON, &keyspace.StrategyOptions)
  385. if err != nil {
  386. return nil, fmt.Errorf(
  387. "Invalid JSON value '%s' as strategy_options for in keyspace '%s': %v",
  388. strategyOptionsJSON, keyspace.Name, err,
  389. )
  390. }
  391. }
  392. return keyspace, nil
  393. }
  394. // query for only the table metadata in the specified keyspace from system.schema_columnfamilies
  395. func getTableMetadata(session *Session, keyspaceName string) ([]TableMetadata, error) {
  396. var (
  397. iter *Iter
  398. scan func(iter *Iter, table *TableMetadata) bool
  399. stmt string
  400. keyAliasesJSON []byte
  401. columnAliasesJSON []byte
  402. )
  403. if session.useSystemSchema { // Cassandra 3.x+
  404. stmt = `
  405. SELECT
  406. table_name
  407. FROM system_schema.tables
  408. WHERE keyspace_name = ?`
  409. switchIter := func() *Iter {
  410. iter.Close()
  411. stmt = `
  412. SELECT
  413. view_name
  414. FROM system_schema.views
  415. WHERE keyspace_name = ?`
  416. iter = session.control.query(stmt, keyspaceName)
  417. return iter
  418. }
  419. scan = func(iter *Iter, table *TableMetadata) bool {
  420. r := iter.Scan(
  421. &table.Name,
  422. )
  423. if !r {
  424. iter = switchIter()
  425. if iter != nil {
  426. switchIter = func() *Iter { return nil }
  427. r = iter.Scan(&table.Name)
  428. }
  429. }
  430. return r
  431. }
  432. } else if session.cfg.ProtoVersion < protoVersion4 {
  433. // we have key aliases
  434. // TODO: Do we need key_aliases?
  435. stmt = `
  436. SELECT
  437. columnfamily_name,
  438. key_validator,
  439. comparator,
  440. default_validator,
  441. key_aliases,
  442. column_aliases,
  443. value_alias
  444. FROM system.schema_columnfamilies
  445. WHERE keyspace_name = ?`
  446. scan = func(iter *Iter, table *TableMetadata) bool {
  447. return iter.Scan(
  448. &table.Name,
  449. &table.KeyValidator,
  450. &table.Comparator,
  451. &table.DefaultValidator,
  452. &keyAliasesJSON,
  453. &columnAliasesJSON,
  454. &table.ValueAlias,
  455. )
  456. }
  457. } else {
  458. stmt = `
  459. SELECT
  460. columnfamily_name,
  461. key_validator,
  462. comparator,
  463. default_validator
  464. FROM system.schema_columnfamilies
  465. WHERE keyspace_name = ?`
  466. scan = func(iter *Iter, table *TableMetadata) bool {
  467. return iter.Scan(
  468. &table.Name,
  469. &table.KeyValidator,
  470. &table.Comparator,
  471. &table.DefaultValidator,
  472. )
  473. }
  474. }
  475. iter = session.control.query(stmt, keyspaceName)
  476. tables := []TableMetadata{}
  477. table := TableMetadata{Keyspace: keyspaceName}
  478. for scan(iter, &table) {
  479. var err error
  480. // decode the key aliases
  481. if keyAliasesJSON != nil {
  482. table.KeyAliases = []string{}
  483. err = json.Unmarshal(keyAliasesJSON, &table.KeyAliases)
  484. if err != nil {
  485. iter.Close()
  486. return nil, fmt.Errorf(
  487. "Invalid JSON value '%s' as key_aliases for in table '%s': %v",
  488. keyAliasesJSON, table.Name, err,
  489. )
  490. }
  491. }
  492. // decode the column aliases
  493. if columnAliasesJSON != nil {
  494. table.ColumnAliases = []string{}
  495. err = json.Unmarshal(columnAliasesJSON, &table.ColumnAliases)
  496. if err != nil {
  497. iter.Close()
  498. return nil, fmt.Errorf(
  499. "Invalid JSON value '%s' as column_aliases for in table '%s': %v",
  500. columnAliasesJSON, table.Name, err,
  501. )
  502. }
  503. }
  504. tables = append(tables, table)
  505. table = TableMetadata{Keyspace: keyspaceName}
  506. }
  507. err := iter.Close()
  508. if err != nil && err != ErrNotFound {
  509. return nil, fmt.Errorf("Error querying table schema: %v", err)
  510. }
  511. return tables, nil
  512. }
  513. func (s *Session) scanColumnMetadataV1(keyspace string) ([]ColumnMetadata, error) {
  514. // V1 does not support the type column, and all returned rows are
  515. // of kind "regular".
  516. const stmt = `
  517. SELECT
  518. columnfamily_name,
  519. column_name,
  520. component_index,
  521. validator,
  522. index_name,
  523. index_type,
  524. index_options
  525. FROM system.schema_columns
  526. WHERE keyspace_name = ?`
  527. var columns []ColumnMetadata
  528. rows := s.control.query(stmt, keyspace).Scanner()
  529. for rows.Next() {
  530. var (
  531. column = ColumnMetadata{Keyspace: keyspace}
  532. indexOptionsJSON []byte
  533. )
  534. // all columns returned by V1 are regular
  535. column.Kind = ColumnRegular
  536. err := rows.Scan(&column.Table,
  537. &column.Name,
  538. &column.ComponentIndex,
  539. &column.Validator,
  540. &column.Index.Name,
  541. &column.Index.Type,
  542. &indexOptionsJSON)
  543. if err != nil {
  544. return nil, err
  545. }
  546. if len(indexOptionsJSON) > 0 {
  547. err := json.Unmarshal(indexOptionsJSON, &column.Index.Options)
  548. if err != nil {
  549. return nil, fmt.Errorf(
  550. "Invalid JSON value '%s' as index_options for column '%s' in table '%s': %v",
  551. indexOptionsJSON,
  552. column.Name,
  553. column.Table,
  554. err)
  555. }
  556. }
  557. columns = append(columns, column)
  558. }
  559. if err := rows.Err(); err != nil {
  560. return nil, err
  561. }
  562. return columns, nil
  563. }
  564. func (s *Session) scanColumnMetadataV2(keyspace string) ([]ColumnMetadata, error) {
  565. // V2+ supports the type column
  566. const stmt = `
  567. SELECT
  568. columnfamily_name,
  569. column_name,
  570. component_index,
  571. validator,
  572. index_name,
  573. index_type,
  574. index_options,
  575. type
  576. FROM system.schema_columns
  577. WHERE keyspace_name = ?`
  578. var columns []ColumnMetadata
  579. rows := s.control.query(stmt, keyspace).Scanner()
  580. for rows.Next() {
  581. var (
  582. column = ColumnMetadata{Keyspace: keyspace}
  583. indexOptionsJSON []byte
  584. )
  585. err := rows.Scan(&column.Table,
  586. &column.Name,
  587. &column.ComponentIndex,
  588. &column.Validator,
  589. &column.Index.Name,
  590. &column.Index.Type,
  591. &indexOptionsJSON,
  592. &column.Kind,
  593. )
  594. if err != nil {
  595. return nil, err
  596. }
  597. if len(indexOptionsJSON) > 0 {
  598. err := json.Unmarshal(indexOptionsJSON, &column.Index.Options)
  599. if err != nil {
  600. return nil, fmt.Errorf(
  601. "Invalid JSON value '%s' as index_options for column '%s' in table '%s': %v",
  602. indexOptionsJSON,
  603. column.Name,
  604. column.Table,
  605. err)
  606. }
  607. }
  608. columns = append(columns, column)
  609. }
  610. if err := rows.Err(); err != nil {
  611. return nil, err
  612. }
  613. return columns, nil
  614. }
  615. func (s *Session) scanColumnMetadataSystem(keyspace string) ([]ColumnMetadata, error) {
  616. const stmt = `
  617. SELECT
  618. table_name,
  619. column_name,
  620. clustering_order,
  621. type,
  622. kind,
  623. position
  624. FROM system_schema.columns
  625. WHERE keyspace_name = ?`
  626. var columns []ColumnMetadata
  627. rows := s.control.query(stmt, keyspace).Scanner()
  628. for rows.Next() {
  629. column := ColumnMetadata{Keyspace: keyspace}
  630. err := rows.Scan(&column.Table,
  631. &column.Name,
  632. &column.ClusteringOrder,
  633. &column.Validator,
  634. &column.Kind,
  635. &column.ComponentIndex,
  636. )
  637. if err != nil {
  638. return nil, err
  639. }
  640. columns = append(columns, column)
  641. }
  642. if err := rows.Err(); err != nil {
  643. return nil, err
  644. }
  645. // TODO(zariel): get column index info from system_schema.indexes
  646. return columns, nil
  647. }
  648. // query for only the column metadata in the specified keyspace from system.schema_columns
  649. func getColumnMetadata(session *Session, keyspaceName string) ([]ColumnMetadata, error) {
  650. var (
  651. columns []ColumnMetadata
  652. err error
  653. )
  654. // Deal with differences in protocol versions
  655. if session.cfg.ProtoVersion == 1 {
  656. columns, err = session.scanColumnMetadataV1(keyspaceName)
  657. } else if session.useSystemSchema { // Cassandra 3.x+
  658. columns, err = session.scanColumnMetadataSystem(keyspaceName)
  659. } else {
  660. columns, err = session.scanColumnMetadataV2(keyspaceName)
  661. }
  662. if err != nil && err != ErrNotFound {
  663. return nil, fmt.Errorf("Error querying column schema: %v", err)
  664. }
  665. return columns, nil
  666. }
  667. // type definition parser state
  668. type typeParser struct {
  669. input string
  670. index int
  671. }
  672. // the type definition parser result
  673. type typeParserResult struct {
  674. isComposite bool
  675. types []TypeInfo
  676. reversed []bool
  677. collections map[string]TypeInfo
  678. }
  679. // Parse the type definition used for validator and comparator schema data
  680. func parseType(def string) typeParserResult {
  681. parser := &typeParser{input: def}
  682. return parser.parse()
  683. }
  684. const (
  685. REVERSED_TYPE = "org.apache.cassandra.db.marshal.ReversedType"
  686. COMPOSITE_TYPE = "org.apache.cassandra.db.marshal.CompositeType"
  687. COLLECTION_TYPE = "org.apache.cassandra.db.marshal.ColumnToCollectionType"
  688. LIST_TYPE = "org.apache.cassandra.db.marshal.ListType"
  689. SET_TYPE = "org.apache.cassandra.db.marshal.SetType"
  690. MAP_TYPE = "org.apache.cassandra.db.marshal.MapType"
  691. )
  692. // represents a class specification in the type def AST
  693. type typeParserClassNode struct {
  694. name string
  695. params []typeParserParamNode
  696. // this is the segment of the input string that defined this node
  697. input string
  698. }
  699. // represents a class parameter in the type def AST
  700. type typeParserParamNode struct {
  701. name *string
  702. class typeParserClassNode
  703. }
  704. func (t *typeParser) parse() typeParserResult {
  705. // parse the AST
  706. ast, ok := t.parseClassNode()
  707. if !ok {
  708. // treat this is a custom type
  709. return typeParserResult{
  710. isComposite: false,
  711. types: []TypeInfo{
  712. NativeType{
  713. typ: TypeCustom,
  714. custom: t.input,
  715. },
  716. },
  717. reversed: []bool{false},
  718. collections: nil,
  719. }
  720. }
  721. // interpret the AST
  722. if strings.HasPrefix(ast.name, COMPOSITE_TYPE) {
  723. count := len(ast.params)
  724. // look for a collections param
  725. last := ast.params[count-1]
  726. collections := map[string]TypeInfo{}
  727. if strings.HasPrefix(last.class.name, COLLECTION_TYPE) {
  728. count--
  729. for _, param := range last.class.params {
  730. // decode the name
  731. var name string
  732. decoded, err := hex.DecodeString(*param.name)
  733. if err != nil {
  734. log.Printf(
  735. "Error parsing type '%s', contains collection name '%s' with an invalid format: %v",
  736. t.input,
  737. *param.name,
  738. err,
  739. )
  740. // just use the provided name
  741. name = *param.name
  742. } else {
  743. name = string(decoded)
  744. }
  745. collections[name] = param.class.asTypeInfo()
  746. }
  747. }
  748. types := make([]TypeInfo, count)
  749. reversed := make([]bool, count)
  750. for i, param := range ast.params[:count] {
  751. class := param.class
  752. reversed[i] = strings.HasPrefix(class.name, REVERSED_TYPE)
  753. if reversed[i] {
  754. class = class.params[0].class
  755. }
  756. types[i] = class.asTypeInfo()
  757. }
  758. return typeParserResult{
  759. isComposite: true,
  760. types: types,
  761. reversed: reversed,
  762. collections: collections,
  763. }
  764. } else {
  765. // not composite, so one type
  766. class := *ast
  767. reversed := strings.HasPrefix(class.name, REVERSED_TYPE)
  768. if reversed {
  769. class = class.params[0].class
  770. }
  771. typeInfo := class.asTypeInfo()
  772. return typeParserResult{
  773. isComposite: false,
  774. types: []TypeInfo{typeInfo},
  775. reversed: []bool{reversed},
  776. }
  777. }
  778. }
  779. func (class *typeParserClassNode) asTypeInfo() TypeInfo {
  780. if strings.HasPrefix(class.name, LIST_TYPE) {
  781. elem := class.params[0].class.asTypeInfo()
  782. return CollectionType{
  783. NativeType: NativeType{
  784. typ: TypeList,
  785. },
  786. Elem: elem,
  787. }
  788. }
  789. if strings.HasPrefix(class.name, SET_TYPE) {
  790. elem := class.params[0].class.asTypeInfo()
  791. return CollectionType{
  792. NativeType: NativeType{
  793. typ: TypeSet,
  794. },
  795. Elem: elem,
  796. }
  797. }
  798. if strings.HasPrefix(class.name, MAP_TYPE) {
  799. key := class.params[0].class.asTypeInfo()
  800. elem := class.params[1].class.asTypeInfo()
  801. return CollectionType{
  802. NativeType: NativeType{
  803. typ: TypeMap,
  804. },
  805. Key: key,
  806. Elem: elem,
  807. }
  808. }
  809. // must be a simple type or custom type
  810. info := NativeType{typ: getApacheCassandraType(class.name)}
  811. if info.typ == TypeCustom {
  812. // add the entire class definition
  813. info.custom = class.input
  814. }
  815. return info
  816. }
  817. // CLASS := ID [ PARAMS ]
  818. func (t *typeParser) parseClassNode() (node *typeParserClassNode, ok bool) {
  819. t.skipWhitespace()
  820. startIndex := t.index
  821. name, ok := t.nextIdentifier()
  822. if !ok {
  823. return nil, false
  824. }
  825. params, ok := t.parseParamNodes()
  826. if !ok {
  827. return nil, false
  828. }
  829. endIndex := t.index
  830. node = &typeParserClassNode{
  831. name: name,
  832. params: params,
  833. input: t.input[startIndex:endIndex],
  834. }
  835. return node, true
  836. }
  837. // PARAMS := "(" PARAM { "," PARAM } ")"
  838. // PARAM := [ PARAM_NAME ":" ] CLASS
  839. // PARAM_NAME := ID
  840. func (t *typeParser) parseParamNodes() (params []typeParserParamNode, ok bool) {
  841. t.skipWhitespace()
  842. // the params are optional
  843. if t.index == len(t.input) || t.input[t.index] != '(' {
  844. return nil, true
  845. }
  846. params = []typeParserParamNode{}
  847. // consume the '('
  848. t.index++
  849. t.skipWhitespace()
  850. for t.input[t.index] != ')' {
  851. // look for a named param, but if no colon, then we want to backup
  852. backupIndex := t.index
  853. // name will be a hex encoded version of a utf-8 string
  854. name, ok := t.nextIdentifier()
  855. if !ok {
  856. return nil, false
  857. }
  858. hasName := true
  859. // TODO handle '=>' used for DynamicCompositeType
  860. t.skipWhitespace()
  861. if t.input[t.index] == ':' {
  862. // there is a name for this parameter
  863. // consume the ':'
  864. t.index++
  865. t.skipWhitespace()
  866. } else {
  867. // no name, backup
  868. hasName = false
  869. t.index = backupIndex
  870. }
  871. // parse the next full parameter
  872. classNode, ok := t.parseClassNode()
  873. if !ok {
  874. return nil, false
  875. }
  876. if hasName {
  877. params = append(
  878. params,
  879. typeParserParamNode{name: &name, class: *classNode},
  880. )
  881. } else {
  882. params = append(
  883. params,
  884. typeParserParamNode{class: *classNode},
  885. )
  886. }
  887. t.skipWhitespace()
  888. if t.input[t.index] == ',' {
  889. // consume the comma
  890. t.index++
  891. t.skipWhitespace()
  892. }
  893. }
  894. // consume the ')'
  895. t.index++
  896. return params, true
  897. }
  898. func (t *typeParser) skipWhitespace() {
  899. for t.index < len(t.input) && isWhitespaceChar(t.input[t.index]) {
  900. t.index++
  901. }
  902. }
  903. func isWhitespaceChar(c byte) bool {
  904. return c == ' ' || c == '\n' || c == '\t'
  905. }
  906. // ID := LETTER { LETTER }
  907. // LETTER := "0"..."9" | "a"..."z" | "A"..."Z" | "-" | "+" | "." | "_" | "&"
  908. func (t *typeParser) nextIdentifier() (id string, found bool) {
  909. startIndex := t.index
  910. for t.index < len(t.input) && isIdentifierChar(t.input[t.index]) {
  911. t.index++
  912. }
  913. if startIndex == t.index {
  914. return "", false
  915. }
  916. return t.input[startIndex:t.index], true
  917. }
  918. func isIdentifierChar(c byte) bool {
  919. return (c >= '0' && c <= '9') ||
  920. (c >= 'a' && c <= 'z') ||
  921. (c >= 'A' && c <= 'Z') ||
  922. c == '-' ||
  923. c == '+' ||
  924. c == '.' ||
  925. c == '_' ||
  926. c == '&'
  927. }