[SDP] [SPARK-52576] Drop/recreate on full refresh and MV update

[sryza]

What changes were proposed in this pull request?

Some pipeline runs result in wiping out and replacing all the data for a table:

• Every run of a materialized view
• Runs of streaming tables that have the "full refresh" flag

In the current implementation, this "wipe out and replace" is implemented by:

• Truncating the table
• Altering the table to drop/update/add columns that don't match the columns in the DataFrame for the current run

The reason that we want originally wanted to truncate + alter instead of drop / recreate is that dropping has some undesirable effects. E.g. it interrupts readers of the table and wipes away things like ACLs.

However, we discovered that not all catalogs support dropping columns (e.g. Hive does not), and there’s no way to tell whether a catalog supports dropping columns or not. So this PR changes the implementation to drop/recreate the table instead of truncate/alter.

Why are the changes needed?

See section above.

Does this PR introduce any user-facing change?

Yes, see section above. No releases contained the old behavior.

How was this patch tested?

• Tests in MaterializeTablesSuite
• Ran the tests in MaterializeTablesSuite with Hive instead of the default catalog

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

No

[szehon-ho]

Makes sense, can change if we add support for drop Column for HMS in the V2SessionCatalog

[szehon-ho]

nit: remove? Optionally add comment about why not truncate/alter?

[gengliangwang]

why do we need this change?

[sryza]

The ordering of columns does not appear to be deterministic (at least across different catalog implementations). Is that unexpected?

[cloud-fan]

for a table, the column order matters. I think we should keep the test as it is and fix the issues we found.

[sryza]

Are we able to get some help with fixing this? What I'm observing is that, with Hive, when I create a table using the following:

      catalog.createTable(
        identifier,
        new TableInfo.Builder()
          .withProperties(mergedProperties.asJava)
          .withColumns(CatalogV2Util.structTypeToV2Columns(outputSchema))
          .withPartitions(partitioning.toArray)
          .build()
      )

and then later fetch the columns using

catalog.loadTable(identifier).columns()

The columns are returned in a different order than they appear in outputSchema.

This happens only with Hive, not the default catalog.

[szehon-ho]

strange, i can take a look

I ran the test case in HiveDDLSuite a few times and cant reproduce it.

    val catalog = spark.sessionState.catalogManager.currentCatalog.asInstanceOf[TableCatalog]
    withTable("t1") {
      val identifier = Identifier.of(Array("default"), "t1")
      val outputSchema = new StructType()
        .add("a", IntegerType, true, "comment1")
        .add("b", IntegerType, true, "comment2")
        .add("c", IntegerType, true, "comment3")
        .add("d", IntegerType, true, "comment4")
      catalog.createTable(
        identifier,
        new TableInfo.Builder()
          .withProperties(Map.empty.asJava)
          .withColumns(CatalogV2Util.structTypeToV2Columns(outputSchema))
          .withPartitions(Array.empty)
          .build()
      )
      val cols = catalog.loadTable(identifier).columns()
      assert(cols.length == 4)
      assert(cols(0).name() == "a")
      assert(cols(1).name() == "b")
      assert(cols(2).name() == "c")
      assert(cols(3).name() == "d")
    }

Is it reproducible with this pr itself?

[cloud-fan]

I think we should fix the Hive catalog to respect the user-specified column order. For now, I'm fine with ignoring some test cases temporarily in the upcoming hive suite, by overriding def excluded from SparkFunSuite, or add column-order methods and override it in the hive suite.

[szehon-ho]

This is going to be a breaking change for HiveCatalog, should we do it off by default and enable via flag ? Looks like some history: https://github.com/apache/spark/blob/master/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala#L820

[szehon-ho]

@cloud-fan @sryza @gengliangwang i made a patch here, does it look like what we want? #51342

[sryza]

OK I've updated this PR to bring back the asserts on the ordering, and we can deal with the Hive tests in the PR that introduces them. @cloud-fan mind taking another look?

[szehon-ho]

Hmm for some reason the original order is working now. Switched it back and now the tests are passing.

@sryza so may not need #51342 after all? :) We can close it then if its not needed, let me know. At least , as part of it, we managed to make the underlying HiveExternalCatalog API better in #51373, that fixes a small issue (SPARK-52681). By the way, it also doesnt prevent you to drop the column and replace them now, can you give a try as well after this pr?

[cloud-fan]

there is a test failure in MaterializeTablesSuite

[sryza]

Hmm for some reason the original order is working now. Switched it back and now the tests are passing.

[cloud-fan]

thanks, merging to master!