Add table name length limit for ingestion

terryyylim · terryyylim · commit 29c73263fc8e · 2021-05-05T17:35:57.000+08:00
Signed-off-by: Terence Lim &lt;terencelimxp@gmail.com&gt;
diff --git a/spark/ingestion/src/main/scala/feast/ingestion/stores/bigtable/BigTableSinkRelation.scala b/spark/ingestion/src/main/scala/feast/ingestion/stores/bigtable/BigTableSinkRelation.scala
@@ -154,7 +154,7 @@ class BigTableSinkRelation(
 
   private def tableName: String = {
     val entities = config.entityColumns.mkString("__")
-    s"${config.projectName}__${entities}"
+    s"${config.projectName}__${entities}".take(maxTableNameLength)
   }
 
   private def joinEntityKey: UserDefinedFunction = udf { r: Row =>
@@ -164,6 +164,7 @@ class BigTableSinkRelation(
   private val metadataColumnFamily = "metadata"
   private val schemaKeyPrefix      = "schema#"
   private val emptyQualifier       = ""
+  private val maxTableNameLength   = 50
 
   private def isSystemColumn(name: String) =
     (config.entityColumns ++ Seq(config.timestampColumn)).contains(name)
diff --git a/spark/ingestion/src/main/scala/feast/ingestion/stores/cassandra/CassandraSinkRelation.scala b/spark/ingestion/src/main/scala/feast/ingestion/stores/cassandra/CassandraSinkRelation.scala
@@ -64,7 +64,7 @@ class CassandraSinkRelation(
   }
 
   def sanitizedForCassandra(expr: String): String = {
-    expr.replace('-', '_')
+    expr.replace('-', '_').take(maxTableNameLength)
   }
 
   val tableName = {
@@ -84,6 +84,8 @@ class CassandraSinkRelation(
 
   val schemaTableName = s"${sparkCatalog}.${keyspace}.feast_schema_reference"
 
+  val maxTableNameLength = 48
+
   def createTable(): Unit = {
 
     sqlContext.sql(s"""

Original file line number	Diff line number	Diff line change
`@@ -64,7 +64,7 @@ class CassandraSinkRelation(`
`64`	`64`	`}`
`65`	`65`
`66`	`66`	`def sanitizedForCassandra(expr: String): String = {`
`67`		`- expr.replace('-', '_')`
	`67`	`+ expr.replace('-', '_').take(maxTableNameLength)`
`68`	`68`	`}`
`69`	`69`
`70`	`70`	`val tableName = {`
`@@ -84,6 +84,8 @@ class CassandraSinkRelation(`
`84`	`84`
`85`	`85`	`val schemaTableName = s"${sparkCatalog}.${keyspace}.feast_schema_reference"`
`86`	`86`
	`87`	`+ val maxTableNameLength = 48`
	`88`	`+`
`87`	`89`	`def createTable(): Unit = {`
`88`	`90`
`89`	`91`	`sqlContext.sql(s"""`