[spark], [infra] run spark integration tests in CI.

[zhongyujiang]

Purpose

The current github CI for Spark module is missing integration tests, and some of Spark's integration tests are actually failing, they've just been consistently ignored by the CI.

Tests

API and Format

Documentation

[zhongyujiang]

cc @Zouxxyy @YannByron Can you please take a look when you have time? Thanks!

[zhongyujiang]

test phase only cover unit tests, not integration tests.

[zhongyujiang]

Some unit tests are failing, it looks like it's releated to the changes in the time travel part. Let me take a look.

[zhongyujiang]

This was changed to throw an exception directly if the time travel fails to find the snapshot.

But when I'm investigating the time travel releated test failures, I found that it seems Paimon currently allows querying a non-existent snapshot and returns an empty result.
Is this behavior reasonable? Shouldn't an exception be thrown if time travel fails to find the snapshot?

paimon/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSourceTest.scala

Lines 89 to 123 in 408b78d

	// set scan.snapshot-id = 4, this query will read data from the next commit.
	val query2 = spark.readStream
	.format("paimon")
	.option("scan.snapshot-id", 4)
	.load(location)
	.writeStream
	.format("memory")
	.option("checkpointLocation", checkpointDir2.getCanonicalPath)
	.queryName("mem_table2")
	.outputMode("append")
	.start()
	val currentResult1 = () => spark.sql("SELECT * FROM mem_table1")
	val currentResult2 = () => spark.sql("SELECT * FROM mem_table2")
	try {
	query1.processAllAvailable()
	query2.processAllAvailable()
	var totalStreamingData1 = latestChanges
	var totalStreamingData2 = Seq.empty[Row]
	checkAnswer(currentResult1(), totalStreamingData1)
	checkAnswer(currentResult2(), totalStreamingData2)
	spark.sql("INSERT INTO T VALUES (40, 'v_40'), (41, 'v_41'), (42, 'v_42')")
	query1.processAllAvailable()
	query2.processAllAvailable()
	totalStreamingData1 =
	totalStreamingData1 ++ (Row(40, "v_40") :: Row(41, "v_41") :: Row(42, "v_42") :: Nil)
	totalStreamingData2 =
	totalStreamingData2 ++ (Row(40, "v_40") :: Row(41, "v_41") :: Row(42, "v_42") :: Nil)
	checkAnswer(currentResult1(), totalStreamingData1)
	checkAnswer(currentResult2(), totalStreamingData2)
	} finally {
	query1.stop()
	query2.stop()
	}