Skip to content

SPARKC-403: Add CLUSTERING ORDER in cql statement #981

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 7 commits into
base: trunk
Choose a base branch
from
Open

SPARKC-403: Add CLUSTERING ORDER in cql statement #981

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
4ae3ff9
Add CLUSTERING ORDER in cql statement
May 17, 2016
5bc7cc6
Add CLUSTERING ORDER in cql statement
May 17, 2016
4dd085c
Changed custom ClusteringOrder class to com.datastax.driver.core.Clus…
Jul 27, 2016
bae67de
Removed list of Clustering order from TableDef
Aug 1, 2016
697dd0e
Changed Options String to Seq[String]
Aug 21, 2016
0aa10be
Changed Options to tableOptions and its type to Map[String,String].
Oct 19, 2016
dfa0382
Changed Options to tableOptions and its type to Map[String,String].
Oct 27, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 41 additions & 4 deletions spark-cassandra-connector/src/main/scala/com/datastax/spark/connector/cql/Schema.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
package com.datastax.spark.connector.cql

import java.io.IOException

import com.datastax.spark.connector._
import com.datastax.spark.connector.mapper.{DataFrameColumnMapper, ColumnMapper}
import org.apache.spark.Logging
Expand All @@ -8,7 +10,6 @@ import org.apache.spark.sql.DataFrame
import scala.collection.JavaConversions._
import scala.language.existentials
import scala.util.{Properties, Try}

import com.datastax.driver.core._
import com.datastax.spark.connector.types.{CounterType, ColumnType}
import com.datastax.spark.connector.util.Quote._
Expand Down Expand Up @@ -92,7 +93,8 @@ case object RegularColumn extends ColumnRole
case class ColumnDef(
columnName: String,
columnRole: ColumnRole,
columnType: ColumnType[_]) extends FieldDef {
columnType: ColumnType[_],
clusteringOrder: ClusteringOrder = ClusteringOrder.ASC) extends FieldDef {

def ref: ColumnRef = ColumnName(columnName)
def isStatic = columnRole == StaticColumn
Expand Down Expand Up @@ -128,6 +130,15 @@ object ColumnDef {
val columnType = ColumnType.fromDriverType(column.getType)
ColumnDef(column.getName, columnRole, columnType)
}

def apply(
column: ColumnMetadata,
columnRole: ColumnRole,
clusteringOrder: ClusteringOrder): ColumnDef = {

val columnType = ColumnType.fromDriverType(column.getType)
ColumnDef(column.getName, columnRole, columnType, clusteringOrder)
}
}

/** A Cassandra table metadata that can be serialized. */
Expand All @@ -138,7 +149,9 @@ case class TableDef(
clusteringColumns: Seq[ColumnDef],
regularColumns: Seq[ColumnDef],
indexes: Seq[IndexDef] = Seq.empty,
isView: Boolean = false) extends StructDef {
isView: Boolean = false,
clusteringOrder: Option[Seq[ClusteringOrder]] = None,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shouldn't this be a property of the ColunDef now? Ie do you really need to specify it separately?

options: String = "") extends StructDef {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I still don't like being able to just pass a string here, If you really think we need this I think it should at least be Seq[String] and we should just require that they not contain "AND" or "WITH". Then we can convert the append code below into a string join instead of having the more complicated logic.

This removes the need for the "appendOptions" function and replaces it with

require(options.forAll( option => !(option.toLowerCase.contains("and") && !(option.toLowerCase.contains("WITH")), "Table options must not contain "WITH OR AND"
(options +: clusteringOptions).mkString("WITH", "AND")```


require(partitionKey.forall(_.isPartitionKeyColumn), "All partition key columns must have role PartitionKeyColumn")
require(clusteringColumns.forall(_.isClusteringColumn), "All clustering columns must have role ClusteringColumn")
Expand Down Expand Up @@ -185,12 +198,36 @@ case class TableDef(
val clusteringColumnNames = clusteringColumns.map(_.columnName).map(quote)
val primaryKeyClause = (partitionKeyClause +: clusteringColumnNames).mkString(", ")

s"""CREATE TABLE ${quote(keyspaceName)}.${quote(tableName)} (
val stmt = s"""CREATE TABLE ${quote(keyspaceName)}.${quote(tableName)} (
| $columnList,
| PRIMARY KEY ($primaryKeyClause)
|)""".stripMargin
val ordered = if (clusteringColumns.nonEmpty)
s"$stmt${Properties.lineSeparator}WITH CLUSTERING ORDER BY (${clusteringColumnOrder(clusteringColumns,clusteringOrder)})"
else stmt
appendOptions(ordered, options)
}
private[this] def clusteringColumnOrder(clusteringColumns: Seq[ColumnDef], clusteringOrder: Option[Seq[ClusteringOrder]]): String = {

val clusteringCols = clusteringOrder match {
case Some(orders) if orders.size == clusteringColumns.size =>
for ( (col, order) <- clusteringColumns zip orders) yield col.copy(clusteringOrder = order)
case Some(e) => throw new IOException("clusteringOrder size is not matching with Clustering Columns")
case None => clusteringColumns
}

clusteringCols.map { col =>
if (col.clusteringOrder == ClusteringOrder.DESC)
s"${quote(col.columnName)} DESC" else s"${quote(col.columnName)} ASC"
}.toList.mkString(", ")
}

private[this] def appendOptions(stmt: String, opts: String) =
if (stmt.contains("WITH") && opts.startsWith("WITH")) s"$stmt${Properties.lineSeparator}AND ${opts.substring(4)}"
else if (!stmt.contains("WITH") && opts.startsWith("AND")) s"WITH ${opts.substring(3)}"
else if (opts == "") s"$stmt"
else s"$stmt${Properties.lineSeparator}$opts"

type ValueRepr = CassandraRow

def newInstance(columnValues: Any*): CassandraRow = {
Expand Down
52 changes: 48 additions & 4 deletions spark-cassandra-connector/src/test/scala/com/datastax/spark/connector/cql/TableDefSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
package com.datastax.spark.connector.cql

import com.datastax.spark.connector.{PartitionKeyColumns, TTL, SomeColumns, AllColumns}
import com.datastax.driver.core.ClusteringOrder
import com.datastax.spark.connector.{AllColumns, PartitionKeyColumns, SomeColumns, TTL}
import com.datastax.spark.connector.types._
import org.scalatest.{WordSpec, Matchers}
import org.scalatest.{Matchers, WordSpec}

class TableDefSpec extends WordSpec with Matchers {

Expand Down Expand Up @@ -32,7 +33,49 @@ class TableDefSpec extends WordSpec with Matchers {
| "c2" varchar,
| "c3" varchar,
| PRIMARY KEY (("c1"), "c2")
|)""".stripMargin
|)
|WITH CLUSTERING ORDER BY ("c2" ASC)""".stripMargin
)
}

"it contains clustering columns with order" in {
val column1 = ColumnDef("c1", PartitionKeyColumn, IntType)
val column2 = ColumnDef("c2", PartitionKeyColumn, VarCharType)
val column3 = ColumnDef("c3", ClusteringColumn(0), VarCharType, ClusteringOrder.DESC)
val column4 = ColumnDef("c4", ClusteringColumn(1), VarCharType)
val column5 = ColumnDef("c5", RegularColumn, VarCharType)
val tableDef = TableDef("keyspace", "table", Seq(column1, column2), Seq(column3, column4), Seq(column5))
tableDef.cql should be(
"""CREATE TABLE "keyspace"."table" (
| "c1" int,
| "c2" varchar,
| "c3" varchar,
| "c4" varchar,
| "c5" varchar,
| PRIMARY KEY (("c1", "c2"), "c3", "c4")
|)
|WITH CLUSTERING ORDER BY ("c3" DESC, "c4" ASC)""".stripMargin
)
}

"it contains clustering columns with order and override order using TableDef" in {
val column1 = ColumnDef("c1", PartitionKeyColumn, IntType)
val column2 = ColumnDef("c2", PartitionKeyColumn, VarCharType)
val column3 = ColumnDef("c3", ClusteringColumn(0), VarCharType, ClusteringOrder.DESC)
val column4 = ColumnDef("c4", ClusteringColumn(1), VarCharType)
val column5 = ColumnDef("c5", RegularColumn, VarCharType)
val tableDef = TableDef("keyspace", "table", Seq(column1, column2), Seq(column3, column4), Seq(column5)
,clusteringOrder=Option(List(ClusteringOrder.ASC,ClusteringOrder.DESC)))
tableDef.cql should be(
"""CREATE TABLE "keyspace"."table" (
| "c1" int,
| "c2" varchar,
| "c3" varchar,
| "c4" varchar,
| "c5" varchar,
| PRIMARY KEY (("c1", "c2"), "c3", "c4")
|)
|WITH CLUSTERING ORDER BY ("c3" ASC, "c4" DESC)""".stripMargin
)
}

Expand All @@ -51,7 +94,8 @@ class TableDefSpec extends WordSpec with Matchers {
| "c4" varchar,
| "c5" varchar,
| PRIMARY KEY (("c1", "c2"), "c3", "c4")
|)""".stripMargin
|)
|WITH CLUSTERING ORDER BY ("c3" ASC, "c4" ASC)""".stripMargin
)
}

Expand Down