[Backport 0.6] Temporarily support 4+ parts table identifier

integ-test/src/integration/scala/org/opensearch/flint/spark/ppl/FlintSparkPPLBasicITSuite.scala

@@ -541,11 +541,6 @@ class FlintSparkPPLBasicITSuite
            | """.stripMargin))
       assert(ex.getMessage().contains("TABLE_OR_VIEW_NOT_FOUND"))
     }
-    val t7 = "spark_catalog.default.flint_ppl_test7.log"
-    val ex = intercept[IllegalArgumentException](sql(s"""
-           | source = $t7| head 2
-           | """.stripMargin))
-    assert(ex.getMessage().contains("Invalid table name"))
   }
 
   test("test describe backtick table names and name contains '.'") {
@@ -564,11 +559,6 @@ class FlintSparkPPLBasicITSuite
            | """.stripMargin))
       assert(ex.getMessage().contains("TABLE_OR_VIEW_NOT_FOUND"))
     }
-    val t7 = "spark_catalog.default.flint_ppl_test7.log"
-    val ex = intercept[IllegalArgumentException](sql(s"""
-           | describe $t7
-           | """.stripMargin))
-    assert(ex.getMessage().contains("Invalid table name"))
   }
 
   test("test explain backtick table names and name contains '.'") {
@@ -590,12 +580,27 @@ class FlintSparkPPLBasicITSuite
         Project(Seq(UnresolvedStar(None)), relation),
         ExplainMode.fromString("extended"))
     comparePlans(logicalPlan, expectedPlan, checkAnalysis = false)
+  }
 
+  // TODO Do not support 4+ parts table identifier in future (may be reverted this PR in 0.8.0)
+  test("test table name with more than 3 parts") {
     val t7 = "spark_catalog.default.flint_ppl_test7.log"
-    val ex = intercept[IllegalArgumentException](sql(s"""
-          | explain extended | source = $t7
-          | """.stripMargin))
-    assert(ex.getMessage().contains("Invalid table name"))
+    val t4Parts = "`spark_catalog`.default.`startTime:1,endTime:2`.`this(is:['a/name'])`"
+    val t5Parts =
+      "`spark_catalog`.default.`startTime:1,endTime:2`.`this(is:['sub/name'])`.`this(is:['sub-sub/name'])`"
+    Seq(t7, t4Parts, t5Parts).foreach { table =>
+      val ex = intercept[AnalysisException](sql(s"""
+                                                   | source = $table| head 2
+                                                   | """.stripMargin))
+      assert(ex.getMessage().contains("TABLE_OR_VIEW_NOT_FOUND"))
+    }
+
+    Seq(t7, t4Parts, t5Parts).foreach { table =>
+      val ex = intercept[AnalysisException](sql(s"""
+                                                   | describe $table
+                                                   | """.stripMargin))
+      assert(ex.getMessage().contains("TABLE_OR_VIEW_NOT_FOUND"))
+    }
   }
 
   test("Search multiple tables - translated into union call with fields") {

ppl-spark-integration/src/main/java/org/opensearch/sql/ppl/utils/RelationUtils.java

@@ -53,8 +53,15 @@ static TableIdentifier getTableIdentifier(QualifiedName qualifiedName) {
                 Option$.MODULE$.apply(qualifiedName.getParts().get(1)),
                 Option$.MODULE$.apply(qualifiedName.getParts().get(0)));
         } else {
-            throw new IllegalArgumentException("Invalid table name: " + qualifiedName
-                + " Syntax: [ database_name. ] table_name");
+            // TODO Do not support 4+ parts table identifier in future (may be reverted this PR in 0.8.0)
+            // qualifiedName.getParts().size() > 3
+            // A Spark TableIdentifier should only contain 3 parts: tableName, databaseName and catalogName.
+            // If the qualifiedName has more than 3 parts,
+            // we merge all parts from 3 to last parts into the tableName as one whole
+            identifier = new TableIdentifier(
+                String.join(".", qualifiedName.getParts().subList(2, qualifiedName.getParts().size())),
+                Option$.MODULE$.apply(qualifiedName.getParts().get(1)),
+                Option$.MODULE$.apply(qualifiedName.getParts().get(0)));
         }
         return identifier;
     }

ppl-spark-integration/src/test/scala/org/opensearch/flint/spark/ppl/PPLLogicalPlanBasicQueriesTranslatorTestSuite.scala

@@ -27,7 +27,8 @@ class PPLLogicalPlanBasicQueriesTranslatorTestSuite
   private val planTransformer = new CatalystQueryPlanVisitor()
   private val pplParser = new PPLSyntaxParser()
 
-  test("test error describe clause") {
+  // TODO Do not support 4+ parts table identifier in future (may be reverted this PR in 0.8.0)
+  ignore("test error describe clause") {
     val context = new CatalystPlanContext
     val thrown = intercept[IllegalArgumentException] {
       planTransformer.visit(plan(pplParser, "describe t.b.c.d"), context)
@@ -50,6 +51,66 @@ class PPLLogicalPlanBasicQueriesTranslatorTestSuite
     comparePlans(expectedPlan, logPlan, false)
   }
 
+  // TODO Do not support 4+ parts table identifier in future (may be reverted this PR in 0.8.0)
+  test("test describe with backticks and more then 3 parts") {
+    val context = new CatalystPlanContext
+    val logPlan =
+      planTransformer.visit(plan(pplParser, "describe `t`.b.`c.d`.`e.f`"), context)
+
+    val expectedPlan = DescribeTableCommand(
+      TableIdentifier("c.d.e.f", Option("b"), Option("t")),
+      Map.empty[String, String].empty,
+      isExtended = true,
+      output = DescribeRelation.getOutputAttrs)
+    comparePlans(expectedPlan, logPlan, false)
+  }
+
+  test("test read table with backticks and more then 3 parts") {
+    val context = new CatalystPlanContext
+    val logPlan =
+      planTransformer.visit(plan(pplParser, "source=`t`.b.`c.d`.`e.f`"), context)
+    val table = UnresolvedRelation(Seq("t", "b", "c.d.e.f"))
+    val expectedPlan = Project(Seq(UnresolvedStar(None)), table)
+    comparePlans(expectedPlan, logPlan, false)
+  }
+
+  test("test describe with complex backticks and more then 3 parts") {
+    val context = new CatalystPlanContext
+    val logPlan =
+      planTransformer.visit(
+        plan(
+          pplParser,
+          "describe `_Basic`.default.`startTime:0,endTime:1`.`logGroups(logGroupIdentifier:['hello/service_log'])`"),
+        context)
+
+    val expectedPlan = DescribeTableCommand(
+      TableIdentifier(
+        "startTime:0,endTime:1.logGroups(logGroupIdentifier:['hello/service_log'])",
+        Option("default"),
+        Option("_Basic")),
+      Map.empty[String, String].empty,
+      isExtended = true,
+      output = DescribeRelation.getOutputAttrs)
+    comparePlans(expectedPlan, logPlan, false)
+  }
+
+  test("test read complex table with backticks and more then 3 parts") {
+    val context = new CatalystPlanContext
+    val logPlan =
+      planTransformer.visit(
+        plan(
+          pplParser,
+          "source=`_Basic`.default.`startTime:0,endTime:1`.`logGroups(logGroupIdentifier:['hello/service_log'])`"),
+        context)
+    val table = UnresolvedRelation(
+      Seq(
+        "_Basic",
+        "default",
+        "startTime:0,endTime:1.logGroups(logGroupIdentifier:['hello/service_log'])"))
+    val expectedPlan = Project(Seq(UnresolvedStar(None)), table)
+    comparePlans(expectedPlan, logPlan, false)
+  }
+
   test("test describe FQN table clause") {
     val context = new CatalystPlanContext
     val logPlan =