[SPARK-46602][SQL] Propagate allowExisting in view creation when the view/table does not exists

[anchovYu]

What changes were proposed in this pull request?

This PR fixes the undesired behavior that concurrent CREATE VIEW IF NOT EXISTS queries could throw TABLE_OR_VIEW_ALREADY_EXISTS exceptions. It's because the current implementation did not propagate the 'IF NOT EXISTS' when the detecting view/table does not exists.

Why are the changes needed?

Fix the above issue.

Does this PR introduce any user-facing change?

Yes in the sense that if fixes an issue in concurrent case.

How was this patch tested?

Without the fix the following test failed while with this PR if passed. But following the comment, I removed the test from this PR.

  test("CREATE VIEW IF NOT EXISTS never throws TABLE_OR_VIEW_ALREADY_EXISTS") {
    // Concurrently create a view with the same name, so that some of the queries may all
    // get that the view does not exist and try to create it. But with IF NOT EXISTS, the
    // queries should not fail.
    import ExecutionContext.Implicits.global
    val concurrency = 10
    val tableName = "table_name"
    val viewName = "view_name"
    withTable(tableName) {
      sql(s"CREATE TABLE $tableName (id int) USING parquet")
      withView("view_name") {
        val futures = (0 to concurrency).map { _ =>
          Future {
            Try {
              sql(s"CREATE VIEW IF NOT EXISTS $viewName AS SELECT * FROM $tableName")
            }
          }
        }
        futures.map { future =>
           val res = ThreadUtils.awaitResult(future, 5.seconds)
           assert(
             res.isSuccess,
             s"Failed to create view: ${if (res.isFailure) res.failed.get.getMessage}"
           )
        }
      }
    }
  }

Was this patch authored or co-authored using generative AI tooling?

No.

[cloud-fan]

the fix is trivial, we don't need to commit this fragile and heavy test. You can just run it offline and verify that the fix works.

[anchovYu]

Yes I have verified that without the fix this test failed.

[cloud-fan]

The failed kafka test is unrelated, thanks, merging to master/3.5!