sunchao commented on code in PR #5935:
URL: https://github.com/apache/datafusion-comet/pull/5935#discussion_r4018560672
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala:
##########
@@ -455,16 +455,135 @@ class CometIcebergWriteDetectionSuite extends
CometTestBase with CometIcebergTes
}
}
- test("Compatible for the remaining supported data location schemes") {
+ test("Compatible when the data location scheme is memory") {
withDetectionCatalog { dir =>
- Seq("gs", "memory").foreach { scheme =>
- val table = s"${scheme}_scheme"
- createTable(
- dir,
- table,
- partitionSpec = "",
- properties =
Some(s"'write.data.path'='$scheme://nonexistent/iceberg/db/$table'"))
- assertSupportLevelIs[Compatible](table, allowWriteFailure = true)
+ createTable(
+ dir,
+ "memory_scheme",
+ partitionSpec = "",
+ properties =
Some("'write.data.path'='memory://nonexistent/iceberg/db/memory_scheme'"))
+ assertSupportLevelIs[Compatible]("memory_scheme", allowWriteFailure =
true)
+ }
+ }
+
+ test("fall-back: gs data location under HadoopFileIO (fs.gs.* is not
forwarded)") {
+ // The hadoop catalog's table.io() is a HadoopFileIO. Planned only, never
executed: running
+ // the write would have the Hadoop GCS connector look for credentials over
the network.
+ withDetectionCatalog { dir =>
+ createTable(
+ dir,
+ "gs_hadoop_io",
+ partitionSpec = "",
+ properties =
Some("'write.data.path'='gs://nonexistent/iceberg/db/gs_hadoop_io'"))
+ assertUnsupportedContains(
+ planInsertWriteExec(s"$catalog.$ns.gs_hadoop_io"),
+ "gs_hadoop_io",
+ "gs://",
+ classOf[HadoopFileIO].getName)
+ }
+ }
+
+ test("gs data location gate decides on the resolved FileIO class") {
+ // Deterministic coverage of every branch; the ResolvingFileIO test below
depends on which
+ // delegate this classpath yields. GCSFileIO is loaded without
initialization so the
+ // optional GCS client libraries are never touched.
+ val location = "gs://bucket/iceberg/db/t"
+ val gcsFileIO =
+ Class.forName(IcebergReflection.ClassNames.GCS_FILE_IO, false,
getClass.getClassLoader)
+ assert(CometIcebergNativeWrite.gcsDataLocationRejection(location,
Some(gcsFileIO)).isEmpty)
+ val hadoop =
+ CometIcebergNativeWrite.gcsDataLocationRejection(location,
Some(classOf[HadoopFileIO]))
+ assert(
+ hadoop.exists(r => r.contains("gs://") &&
r.contains(classOf[HadoopFileIO].getName)),
+ hadoop)
+ val unresolved =
CometIcebergNativeWrite.gcsDataLocationRejection(location, None)
+ assert(unresolved.exists(_.contains("gs://")), unresolved)
+ }
+
+ test("fall-back: gs data location under ResolvingFileIO whose GCSFileIO
fails to initialize") {
+ // ResolvingFileIO.ioClass maps gs:// to GCSFileIO, but the delegate it
instantiates is a
+ // HadoopFileIO whenever loading or initializing GCSFileIO throws an
IllegalArgumentException,
+ // so the gate must judge the instantiated delegate. An unparseable GCS
chunk size makes
+ // GCSFileIO.initialize throw NumberFormatException where the GCS client
libraries are
+ // present; where they are absent, loading fails earlier (or, when only
some of them are
+ // present, construction fails with an error Iceberg does not fall back
from, and the writer
+ // itself would fail). On every classpath the effective delegate is never
GCSFileIO and the
+ // write must be declined.
+ withTempIcebergDir { warehouseDir =>
+ val location = "gs://nonexistent/iceberg/db/gs_resolving_bad"
+ val badProperty = "gcs.channel.read.chunk-size-bytes" -> "invalid"
+ val resolving = new ResolvingFileIO()
+ resolving.setConf(new Configuration())
+ resolving.initialize(java.util.Collections.singletonMap(badProperty._1,
badProperty._2))
+ val delegate =
+ try IcebergReflection.resolveFileIOClass(resolving, location)
+ finally resolving.close()
+ logInfo(s"ResolvingFileIO delegate with $badProperty on this classpath:
$delegate")
+ assert(delegate.forall(_ == classOf[HadoopFileIO]), delegate)
Review Comment:
[P2] Make the initialization-failure fixture version-aware
### Correctness
With Iceberg 1.10.0 or 1.11.0 and the dependencies needed to construct
GCSFileIO, this assertion fails: `GCSFileIO.initialize` only stores the
properties and initializes metrics ([1.10
source](https://github.com/apache/iceberg/blob/2114bf631e49af532d66e2ce148ee49dd1dd1f1f/gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSFileIO.java#L170-L173),
[1.11
source](https://github.com/apache/iceberg/blob/6976e020b894f6a6777704df2b8c4458cb291ae9/gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSFileIO.java#L164-L167)).
Parsing the invalid chunk size is deferred until client construction. I
reproduced this with both real Iceberg runtimes and the exact current
reflection helper: it returns `Some(GCSFileIO)`, so `delegate.forall(_ ==
classOf[HadoopFileIO])` is false. The test can currently pass through an
earlier missing-class failure/None instead of the intended initialization
fallback. Please use a fixture that actually fails during delegate
initialization on the tested version, or scope this p
roperty-based case to the versions where it does so and preserve the GCS
delegate expectation on newer versions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
