sunchao commented on code in PR #6441:
URL: https://github.com/apache/datafusion-comet/pull/6441#discussion_r4168266078
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:
##########
@@ -165,6 +280,52 @@ object CometIcebergNativeWrite extends
CometOperatorSerde[IcebergWriteExec] {
private type TriggerRule = TriggerContext => Option[String]
+ private def effectiveHadoopConf(op: IcebergWriteExec, table: Any):
Configuration = {
+ val sessionConf = op.session.sessionState.newHadoopConf()
+ val catalogOverrides = IcebergReflection
+ .getTableName(table)
+ .flatMap(_.split("\\.", 2).headOption)
+ .map { catalogName =>
+ val prefix = s"spark.sql.catalog.$catalogName.hadoop."
+ op.session.sessionState.conf.getAllConfs.collect {
+ case (key, value) if key.startsWith(prefix) =>
key.substring(prefix.length) -> value
+ }
+ }
+ .getOrElse(Map.empty)
+
+ IcebergReflection.getFileIOHadoopConf(table) match {
+ case None =>
+ catalogOverrides.foreach { case (key, value) => sessionConf.set(key,
value) }
+ sessionConf
+ case Some(fileIOConf) =>
+ // HadoopFileIO stores its configuration in Iceberg's
SerializableConfiguration. That
+ // class rebuilds a Configuration(false) by calling set() for every
entry, so Hadoop's
+ // property-source metadata is lost and core-default.xml values look
programmatic. Keep
+ // the current session configuration as the base so built-in defaults
retain their
+ // provenance and write-time settings are not replaced by stale FileIO
defaults. Retain
+ // other FileIO-only values, then apply SparkCatalog's explicit
catalog hadoop.* overrides
+ // last because Iceberg gives them precedence over the session
configuration.
+ val effectiveConf = new Configuration(sessionConf)
+ fileIOConf.iterator().asScala.foreach { entry =>
+ val key = entry.getKey
+ val fileIOValue = fileIOConf.getRaw(key)
+ if (!catalogOverrides.contains(key) &&
+ !hasExplicitSource(sessionConf, key) &&
+ fileIOValue != sessionConf.getRaw(key)) {
+ effectiveConf.set(key, fileIOValue)
+ }
+ }
+ catalogOverrides.foreach { case (key, value) => effectiveConf.set(key,
value) }
Review Comment:
[P2] Could we preserve the initialized FileIO's effective values instead of
replaying current catalog options here? Initialize `cat` with
`spark.hadoop.fs.s3a.endpoint=https://original.example.test`, then set
`spark.sql.catalog.cat.hadoop.fs.s3a.endpoint=https://changed.example.test`
before an insert. SparkCatalog retains its original HadoopFileIO configuration,
but this loop changes the endpoint serialized for native writes. The base and
JVM writer both retain the original endpoint. Native output now targets
different storage while JVM footer reads and cleanup still use the original
FileIO, breaking previously valid writes. Use the FileIO snapshot as the
authority for values, recover default-source metadata separately, and add a
regression covering changes after catalog initialization.
Evidence: A bounded local reproduction used Spark 4.1.3, Iceberg 1.11.0, a
Hadoop catalog with local metadata and an S3 data location, and the exact-head
configuration/translation helpers. After changing the catalog endpoint,
reloading the table still returned FileIO endpoint
`https://original.example.test`. Base translation returned that same endpoint.
Head translation returned `https://changed.example.test`, and the new gate
returned no rejected keys. SparkCatalog initializes its Hadoop configuration
once. `CometIcebergWriteExec` subsequently passes the original `tableIO` to
footer reconstruction and cleanup. No S3 requests were made.
--
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]
