philo-he commented on code in PR #12549:
URL: https://github.com/apache/gluten/pull/12549#discussion_r3801867600
##########
backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala:
##########
@@ -845,7 +862,8 @@ object VeloxConfig extends ConfigRegistry {
.createWithDefault(50)
val CUDF_ENABLE_TABLE_SCAN =
-
buildStaticConf("spark.gluten.sql.columnar.backend.velox.cudf.enableTableScan")
Review Comment:
It appears that this config is used at backend init time. Should we keep it
static?
##########
gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala:
##########
@@ -474,119 +473,154 @@ object GlutenConfig extends ConfigRegistry {
val SPARK_UNSAFE_SORTER_SPILL_READER_BUFFER_SIZE =
"spark.unsafe.sorter.spill.reader.buffer.size"
val SPARK_SHUFFLE_SPILL_DISK_WRITE_BUFFER_SIZE =
"spark.shuffle.spill.diskWriteBufferSize"
val SPARK_SHUFFLE_SPILL_COMPRESS = "spark.shuffle.spill.compress"
- val SPARK_SHUFFLE_SPILL_COMPRESS_DEFAULT: Boolean = true
+ // The codec `spark.gluten.sql.columnar.shuffle.codec` falls back to, and
its Spark default.
+ val SPARK_IO_COMPRESSION_CODEC = "spark.io.compression.codec"
+ val SPARK_IO_COMPRESSION_CODEC_DEFAULT = "lz4"
val SPARK_MAX_BROADCAST_TABLE_SIZE = "spark.sql.maxBroadcastTableSize"
- def get: GlutenConfig = {
+ override def get: GlutenConfig = {
new GlutenConfig(GlutenCoreConfig.activeSQLConf)
}
def prefixOf(backendName: String): String =
s"spark.gluten.sql.columnar.backend.$backendName"
def prefixSessionOf(backendName: String): String =
s"spark.gluten.$backendName"
- private lazy val nativeKeys = Set(
- DEBUG_ENABLED.key,
- BENCHMARK_SAVE_DIR.key,
- GlutenCoreConfig.COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES.key,
- COLUMNAR_MAX_BATCH_SIZE.key,
- SHUFFLE_WRITER_BUFFER_SIZE.key,
- COLUMNAR_CUDF_ENABLED.key,
- SQLConf.LEGACY_SIZE_OF_NULL.key,
- SQLConf.LEGACY_STATISTICAL_AGGREGATE.key,
- SQLConf.JSON_GENERATOR_IGNORE_NULL_FIELDS.key,
- SQLConf.RUNTIME_BLOOM_FILTER_EXPECTED_NUM_ITEMS.key,
- SQLConf.RUNTIME_BLOOM_FILTER_NUM_BITS.key,
- SQLConf.RUNTIME_BLOOM_FILTER_MAX_NUM_BITS.key,
- SQLConf.RUNTIME_BLOOM_FILTER_MAX_NUM_ITEMS.key,
- "spark.io.compression.codec",
- "spark.sql.decimalOperations.allowPrecisionLoss",
- // s3 config
- SPARK_S3_ACCESS_KEY,
- SPARK_S3_SECRET_KEY,
- SPARK_S3_ENDPOINT,
- SPARK_S3_CONNECTION_SSL_ENABLED,
- SPARK_S3_PATH_STYLE_ACCESS,
- SPARK_S3_USE_INSTANCE_CREDENTIALS,
- SPARK_S3_IAM,
- SPARK_S3_IAM_SESSION_NAME,
- SPARK_S3_RETRY_MAX_ATTEMPTS,
- SPARK_S3_CONNECTION_MAXIMUM,
- SPARK_S3_ENDPOINT_REGION,
- SPARK_S3_AWS_IMDS_ENABLED,
- "spark.gluten.velox.fs.s3a.retry.mode",
- "spark.gluten.velox.awsSdkLogLevel",
- "spark.gluten.velox.s3UseProxyFromEnv",
- "spark.gluten.velox.s3PayloadSigningPolicy",
- "spark.gluten.velox.s3LogLocation",
Review Comment:
Do we need to keep this config? It seems that it is still used by Velox.
There are several configs documented in the below link. Could you
double-check them? If we indeed should some of them, please help make the
document aligned. Thanks!
https://github.com/apache/gluten/blob/8439151d606e41abd9a6e91e17e887a4a7a7dc9b/docs/get-started/VeloxS3.md?plain=1#L194-L201
##########
backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala:
##########
@@ -641,23 +644,37 @@ object VeloxConfig extends ConfigRegistry {
val CACHE_PREFETCH_MINPCT =
buildStaticConf("spark.gluten.sql.columnar.backend.velox.cachePrefetchMinPct")
+ .passToNative()
.doc("Set prefetch cache min pct for velox file scan")
.intConf
.createWithDefault(0)
+ // The three confs below are read by native `createHiveConnectorConfig`,
which runs both at
+ // backend initialization (for the reused HiveConnector) and per write on
each native runtime
+ // (`VeloxParquetDataSourceS3::initSink`, `IcebergWriter`), hence they are
modifiable confs.
val AWS_SDK_LOG_LEVEL =
buildConf("spark.gluten.velox.awsSdkLogLevel")
.internal()
+ .passToNative()
.doc("Log granularity of AWS C++ SDK in velox.")
.stringConf
.createWithDefault("FATAL")
- val AWS_S3_RETRY_MODE =
- buildConf("spark.gluten.velox.fs.s3a.retry.mode")
+ val S3_USE_PROXY_FROM_ENV =
+ buildConf("spark.gluten.velox.s3UseProxyFromEnv")
.internal()
- .doc("Retry mode for AWS s3 connection error: legacy, standard and
adaptive.")
+ .passToNative()
+ .doc("Whether to use proxy from environment variables for S3 C++
client.")
+ .booleanConf
+ .createWithDefault(false)
+
+ val S3_PAYLOAD_SIGNING_POLICY =
+ buildConf("spark.gluten.velox.s3PayloadSigningPolicy")
+ .internal()
+ .passToNative()
Review Comment:
Not sure if this a valid concern.
What if developers forget to add `passToNative` for a new config? Will the
default value defined in native be used? If it makes sense, perhaps, we need
somehow avoid this unintentional behavior in a separate PR.
##########
gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala:
##########
@@ -146,9 +146,9 @@ private object GlutenDriverPlugin extends Logging {
// Set off-heap size in bytes per task.
val taskSlots = SparkResourceUtil.getTaskSlots(conf)
- conf.set(GlutenCoreConfig.NUM_TASK_SLOTS_PER_EXECUTOR, taskSlots)
+ conf.set(GlutenCoreConfig.NUM_TASK_SLOTS_PER_EXECUTOR.key,
taskSlots.toString)
val offHeapPerTask = offHeapSize / taskSlots
- conf.set(GlutenCoreConfig.COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES,
offHeapPerTask)
+ conf.set(GlutenCoreConfig.COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES.key,
offHeapPerTask.toString)
Review Comment:
Do we have to change these two lines? It seems that they are equivalent.
--
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]
