Add table name length limit for ingestion

terryyylim · terryyylim · commit 34d2dc3463cf · 2021-05-05T14:57:32.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
@@ -65,6 +65,7 @@ class CassandraSinkRelation(
 
   def sanitizedForCassandra(expr: String): String = {
     expr.replace('-', '_')
+    expr.take(maxTableNameLength)
   }
 
   val tableName = {
@@ -84,6 +85,8 @@ class CassandraSinkRelation(
 
   val schemaTableName = s"${sparkCatalog}.${keyspace}.feast_schema_reference"
 
+  val maxTableNameLength = 48
+
   def createTable(): Unit = {
 
     sqlContext.sql(s"""
