| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| package org.apache.comet |
| |
| import java.util.Locale |
| import java.util.concurrent.TimeUnit |
| |
| import scala.collection.mutable.ListBuffer |
| |
| import org.apache.spark.network.util.ByteUnit |
| import org.apache.spark.network.util.JavaUtils |
| import org.apache.spark.sql.comet.util.Utils |
| import org.apache.spark.sql.internal.SQLConf |
| |
| import org.apache.comet.shims.ShimCometConf |
| |
| /** |
| * Configurations for a Comet application. Mostly inspired by [[SQLConf]] in Spark. |
| * |
| * To get the value of a Comet config key from a [[SQLConf]], you can do the following: |
| * |
| * {{{ |
| * CometConf.COMET_ENABLED.get |
| * }}} |
| * |
| * which retrieves the config value from the thread-local [[SQLConf]] object. Alternatively, you |
| * can also explicitly pass a [[SQLConf]] object to the `get` method. |
| */ |
| object CometConf extends ShimCometConf { |
| |
| val COMPAT_GUIDE: String = "For more information, refer to the Comet Compatibility " + |
| "Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html)" |
| |
| private val TUNING_GUIDE = "For more information, refer to the Comet Tuning " + |
| "Guide (https://datafusion.apache.org/comet/user-guide/tuning.html)" |
| |
| private val TRACING_GUIDE = "For more information, refer to the Comet Tracing " + |
| "Guide (https://datafusion.apache.org/comet/contributor-guide/tracing.html)" |
| |
| private val DEBUGGING_GUIDE = "For more information, refer to the Comet Debugging " + |
| "Guide (https://datafusion.apache.org/comet/contributor-guide/debugging.html)" |
| |
| /** List of all configs that is used for generating documentation */ |
| val allConfs = new ListBuffer[ConfigEntry[_]] |
| |
| private val CATEGORY_SCAN = "scan" |
| private val CATEGORY_PARQUET = "parquet" |
| private val CATEGORY_EXEC = "exec" |
| private val CATEGORY_EXEC_EXPLAIN = "exec_explain" |
| private val CATEGORY_ENABLE_EXEC = "enable_exec" |
| private val CATEGORY_SHUFFLE = "shuffle" |
| private val CATEGORY_TUNING = "tuning" |
| private val CATEGORY_TESTING = "testing" |
| |
| def register(conf: ConfigEntry[_]): Unit = { |
| assert(conf.category.nonEmpty, s"${conf.key} does not have a category defined") |
| allConfs.append(conf) |
| } |
| |
| def conf(key: String): ConfigBuilder = ConfigBuilder(key) |
| |
| val COMET_PREFIX = "spark.comet"; |
| |
| val COMET_EXEC_CONFIG_PREFIX: String = s"$COMET_PREFIX.exec" |
| |
| val COMET_EXPR_CONFIG_PREFIX: String = s"$COMET_PREFIX.expression" |
| |
| val COMET_OPERATOR_CONFIG_PREFIX: String = s"$COMET_PREFIX.operator" |
| |
| val COMET_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.enabled") |
| .category(CATEGORY_EXEC) |
| .doc( |
| "Whether to enable Comet extension for Spark. When this is turned on, Spark will use " + |
| "Comet to read Parquet data source. Note that to enable native vectorized execution, " + |
| "both this config and `spark.comet.exec.enabled` need to be enabled.") |
| .booleanConf |
| .createWithEnvVarOrDefault("ENABLE_COMET", true) |
| |
| val COMET_NATIVE_SCAN_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.scan.enabled") |
| .category(CATEGORY_SCAN) |
| .doc( |
| "Whether to enable native scans. When this is turned on, Spark will use Comet to " + |
| "read supported data sources (currently only Parquet is supported natively). Note " + |
| "that to enable native vectorized execution, both this config and " + |
| "`spark.comet.exec.enabled` need to be enabled.") |
| .booleanConf |
| .createWithDefault(true) |
| |
| val COMET_NATIVE_PARQUET_WRITE_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.parquet.write.enabled") |
| .category(CATEGORY_TESTING) |
| .doc( |
| "Whether to enable native Parquet write through Comet. When enabled, " + |
| "Comet will intercept Parquet write operations and execute them natively. This " + |
| "feature is highly experimental and only partially implemented. It should not " + |
| "be used in production.") |
| .booleanConf |
| .createWithEnvVarOrDefault("ENABLE_COMET_WRITE", false) |
| |
| // Deprecated: native_comet uses mutable buffers incompatible with Arrow FFI best practices |
| // and does not support complex types. Use native_iceberg_compat or auto instead. |
| // This will be removed in a future release. |
| // See: https://github.com/apache/datafusion-comet/issues/2186 |
| @deprecated("Use SCAN_AUTO instead. native_comet will be removed in a future release.", "0.9.0") |
| val SCAN_NATIVE_COMET = "native_comet" |
| val SCAN_NATIVE_DATAFUSION = "native_datafusion" |
| val SCAN_NATIVE_ICEBERG_COMPAT = "native_iceberg_compat" |
| val SCAN_AUTO = "auto" |
| |
| val COMET_NATIVE_SCAN_IMPL: ConfigEntry[String] = conf("spark.comet.scan.impl") |
| .category(CATEGORY_PARQUET) |
| .doc( |
| "The implementation of Comet's Parquet scan to use. Available scans are " + |
| s"`$SCAN_NATIVE_DATAFUSION`, and `$SCAN_NATIVE_ICEBERG_COMPAT`. " + |
| s"`$SCAN_NATIVE_DATAFUSION` is a fully native implementation, and " + |
| s"`$SCAN_NATIVE_ICEBERG_COMPAT` is a hybrid implementation that supports some " + |
| "additional features, such as row indexes and field ids. " + |
| s"`$SCAN_AUTO` (default) chooses the best available scan based on the scan schema.") |
| .stringConf |
| .transform(_.toLowerCase(Locale.ROOT)) |
| .checkValues(Set(SCAN_NATIVE_DATAFUSION, SCAN_NATIVE_ICEBERG_COMPAT, SCAN_AUTO)) |
| .createWithEnvVarOrDefault("COMET_PARQUET_SCAN_IMPL", SCAN_AUTO) |
| |
| val COMET_ICEBERG_NATIVE_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.scan.icebergNative.enabled") |
| .category(CATEGORY_SCAN) |
| .doc( |
| "Whether to enable native Iceberg table scan using iceberg-rust. When enabled, " + |
| "Iceberg tables are read directly through native execution, bypassing Spark's " + |
| "DataSource V2 API for better performance.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_ICEBERG_DATA_FILE_CONCURRENCY_LIMIT: ConfigEntry[Int] = |
| conf("spark.comet.scan.icebergNative.dataFileConcurrencyLimit") |
| .category(CATEGORY_SCAN) |
| .doc( |
| "The number of Iceberg data files to read concurrently within a single task. " + |
| "Higher values improve throughput for tables with many small files by overlapping " + |
| "I/O latency, but increase memory usage. Values between 2 and 8 are suggested.") |
| .intConf |
| .checkValue(v => v > 0, "Data file concurrency limit must be positive") |
| .createWithDefault(1) |
| |
| val COMET_CSV_V2_NATIVE_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.scan.csv.v2.enabled") |
| .category(CATEGORY_TESTING) |
| .doc( |
| "Whether to use the native Comet V2 CSV reader for improved performance. " + |
| "Default: false (uses standard Spark CSV reader) " + |
| "Experimental: Performance benefits are workload-dependent.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_RESPECT_PARQUET_FILTER_PUSHDOWN: ConfigEntry[Boolean] = |
| conf("spark.comet.parquet.respectFilterPushdown") |
| .category(CATEGORY_PARQUET) |
| .doc( |
| "Whether to respect Spark's PARQUET_FILTER_PUSHDOWN_ENABLED config. This needs to be " + |
| "respected when running the Spark SQL test suite but the default setting " + |
| "results in poor performance in Comet when using the new native scans, " + |
| "disabled by default") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_PARQUET_PARALLEL_IO_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.parquet.read.parallel.io.enabled") |
| .category(CATEGORY_PARQUET) |
| .doc( |
| "Whether to enable Comet's parallel reader for Parquet files. The parallel reader reads " + |
| "ranges of consecutive data in a file in parallel. It is faster for large files and " + |
| "row groups but uses more resources.") |
| .booleanConf |
| .createWithDefault(true) |
| |
| val COMET_PARQUET_PARALLEL_IO_THREADS: ConfigEntry[Int] = |
| conf("spark.comet.parquet.read.parallel.io.thread-pool.size") |
| .category(CATEGORY_PARQUET) |
| .doc("The maximum number of parallel threads the parallel reader will use in a single " + |
| "executor. For executors configured with a smaller number of cores, use a smaller number.") |
| .intConf |
| .createWithDefault(16) |
| |
| val COMET_IO_MERGE_RANGES: ConfigEntry[Boolean] = |
| conf("spark.comet.parquet.read.io.mergeRanges") |
| .category(CATEGORY_PARQUET) |
| .doc( |
| "When enabled the parallel reader will try to merge ranges of data that are separated " + |
| "by less than `comet.parquet.read.io.mergeRanges.delta` bytes. Longer continuous reads " + |
| "are faster on cloud storage.") |
| .booleanConf |
| .createWithDefault(true) |
| |
| val COMET_IO_MERGE_RANGES_DELTA: ConfigEntry[Int] = |
| conf("spark.comet.parquet.read.io.mergeRanges.delta") |
| .category(CATEGORY_PARQUET) |
| .doc("The delta in bytes between consecutive read ranges below which the parallel reader " + |
| "will try to merge the ranges. The default is 8MB.") |
| .intConf |
| .createWithDefault(1 << 23) // 8 MB |
| |
| val COMET_IO_ADJUST_READRANGE_SKEW: ConfigEntry[Boolean] = |
| conf("spark.comet.parquet.read.io.adjust.readRange.skew") |
| .category(CATEGORY_PARQUET) |
| .doc("In the parallel reader, if the read ranges submitted are skewed in sizes, this " + |
| "option will cause the reader to break up larger read ranges into smaller ranges to " + |
| "reduce the skew. This will result in a slightly larger number of connections opened to " + |
| "the file system but may give improved performance.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_CONVERT_FROM_PARQUET_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.convert.parquet.enabled") |
| .category(CATEGORY_TESTING) |
| .doc( |
| "When enabled, data from Spark (non-native) Parquet v1 and v2 scans will be converted to " + |
| "Arrow format. This is an experimental feature and has known issues with " + |
| "non-UTC timezones.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_CONVERT_FROM_JSON_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.convert.json.enabled") |
| .category(CATEGORY_TESTING) |
| .doc( |
| "When enabled, data from Spark (non-native) JSON v1 and v2 scans will be converted to " + |
| "Arrow format. This is an experimental feature and has known issues with " + |
| "non-UTC timezones.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_CONVERT_FROM_CSV_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.convert.csv.enabled") |
| .category(CATEGORY_TESTING) |
| .doc( |
| "When enabled, data from Spark (non-native) CSV v1 and v2 scans will be converted to " + |
| "Arrow format. This is an experimental feature and has known issues with " + |
| "non-UTC timezones.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_EXEC_ENABLED: ConfigEntry[Boolean] = conf(s"$COMET_EXEC_CONFIG_PREFIX.enabled") |
| .category(CATEGORY_EXEC) |
| .doc( |
| "Whether to enable Comet native vectorized execution for Spark. This controls whether " + |
| "Spark should convert operators into their Comet counterparts and execute them in " + |
| "native space. Note: each operator is associated with a separate config in the " + |
| "format of `spark.comet.exec.<operator_name>.enabled` at the moment, and both the " + |
| "config and this need to be turned on, in order for the operator to be executed in " + |
| "native.") |
| .booleanConf |
| .createWithDefault(true) |
| |
| val COMET_EXEC_PROJECT_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("project", defaultValue = true) |
| val COMET_EXEC_FILTER_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("filter", defaultValue = true) |
| val COMET_EXEC_SORT_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("sort", defaultValue = true) |
| val COMET_EXEC_LOCAL_LIMIT_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("localLimit", defaultValue = true) |
| val COMET_EXEC_GLOBAL_LIMIT_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("globalLimit", defaultValue = true) |
| val COMET_EXEC_BROADCAST_HASH_JOIN_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("broadcastHashJoin", defaultValue = true) |
| val COMET_EXEC_BROADCAST_EXCHANGE_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("broadcastExchange", defaultValue = true) |
| val COMET_EXEC_HASH_JOIN_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("hashJoin", defaultValue = true) |
| val COMET_EXEC_SORT_MERGE_JOIN_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("sortMergeJoin", defaultValue = true) |
| val COMET_EXEC_AGGREGATE_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("aggregate", defaultValue = true) |
| val COMET_EXEC_COLLECT_LIMIT_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("collectLimit", defaultValue = true) |
| val COMET_EXEC_COALESCE_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("coalesce", defaultValue = true) |
| val COMET_EXEC_UNION_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("union", defaultValue = true) |
| val COMET_EXEC_EXPAND_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("expand", defaultValue = true) |
| val COMET_EXEC_EXPLODE_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("explode", defaultValue = true) |
| val COMET_EXEC_WINDOW_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("window", defaultValue = true) |
| val COMET_EXEC_TAKE_ORDERED_AND_PROJECT_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("takeOrderedAndProject", defaultValue = true) |
| val COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED: ConfigEntry[Boolean] = |
| createExecEnabledConfig("localTableScan", defaultValue = false) |
| |
| val COMET_NATIVE_COLUMNAR_TO_ROW_ENABLED: ConfigEntry[Boolean] = |
| conf(s"$COMET_EXEC_CONFIG_PREFIX.columnarToRow.native.enabled") |
| .category(CATEGORY_EXEC) |
| .doc( |
| "Whether to enable native columnar to row conversion. When enabled, Comet will use " + |
| "native Rust code to convert Arrow columnar data to Spark UnsafeRow format instead " + |
| "of the JVM implementation. This can improve performance for queries that need to " + |
| "convert between columnar and row formats. This is an experimental feature.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_EXEC_SORT_MERGE_JOIN_WITH_JOIN_FILTER_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.exec.sortMergeJoinWithJoinFilter.enabled") |
| .category(CATEGORY_ENABLE_EXEC) |
| .doc("Experimental support for Sort Merge Join with filter") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_TRACING_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.tracing.enabled") |
| .category(CATEGORY_TUNING) |
| .doc(s"Enable fine-grained tracing of events and memory usage. $TRACING_GUIDE.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_ONHEAP_MEMORY_OVERHEAD: ConfigEntry[Long] = conf("spark.comet.memoryOverhead") |
| .category(CATEGORY_TESTING) |
| .doc( |
| "The amount of additional memory to be allocated per executor process for Comet, in MiB, " + |
| "when running Spark in on-heap mode.") |
| .bytesConf(ByteUnit.MiB) |
| .createWithDefault(1024) |
| |
| val COMET_EXEC_SHUFFLE_ENABLED: ConfigEntry[Boolean] = |
| conf(s"$COMET_EXEC_CONFIG_PREFIX.shuffle.enabled") |
| .category(CATEGORY_SHUFFLE) |
| .doc( |
| "Whether to enable Comet native shuffle. " + |
| "Note that this requires setting `spark.shuffle.manager` to " + |
| "`org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager`. " + |
| "`spark.shuffle.manager` must be set before starting the Spark application and " + |
| "cannot be changed during the application.") |
| .booleanConf |
| .createWithDefault(true) |
| |
| val COMET_SHUFFLE_MODE: ConfigEntry[String] = conf(s"$COMET_EXEC_CONFIG_PREFIX.shuffle.mode") |
| .category(CATEGORY_SHUFFLE) |
| .doc( |
| "This is test config to allow tests to force a particular shuffle implementation to be " + |
| "used. Valid values are `jvm` for Columnar Shuffle, `native` for Native Shuffle, " + |
| s"and `auto` to pick the best supported option (`native` has priority). $TUNING_GUIDE.") |
| .internal() |
| .stringConf |
| .transform(_.toLowerCase(Locale.ROOT)) |
| .checkValues(Set("native", "jvm", "auto")) |
| .createWithDefault("auto") |
| |
| val COMET_EXEC_BROADCAST_FORCE_ENABLED: ConfigEntry[Boolean] = |
| conf(s"$COMET_EXEC_CONFIG_PREFIX.broadcast.enabled") |
| .category(CATEGORY_EXEC) |
| .doc( |
| "Whether to force enabling broadcasting for Comet native operators. " + |
| "Comet broadcast feature will be enabled automatically by " + |
| "Comet extension. But for unit tests, we need this feature to force enabling it " + |
| "for invalid cases. So this config is only used for unit test.") |
| .internal() |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_REPLACE_SMJ: ConfigEntry[Boolean] = |
| conf(s"$COMET_EXEC_CONFIG_PREFIX.replaceSortMergeJoin") |
| .category(CATEGORY_EXEC) |
| .doc("Experimental feature to force Spark to replace SortMergeJoin with ShuffledHashJoin " + |
| s"for improved performance. This feature is not stable yet. $TUNING_GUIDE.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_EXEC_SHUFFLE_WITH_HASH_PARTITIONING_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.native.shuffle.partitioning.hash.enabled") |
| .category(CATEGORY_SHUFFLE) |
| .doc("Whether to enable hash partitioning for Comet native shuffle.") |
| .booleanConf |
| .createWithDefault(true) |
| |
| val COMET_EXEC_SHUFFLE_WITH_RANGE_PARTITIONING_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.native.shuffle.partitioning.range.enabled") |
| .category(CATEGORY_SHUFFLE) |
| .doc("Whether to enable range partitioning for Comet native shuffle.") |
| .booleanConf |
| .createWithDefault(true) |
| |
| val COMET_EXEC_SHUFFLE_WITH_ROUND_ROBIN_PARTITIONING_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.native.shuffle.partitioning.roundrobin.enabled") |
| .category(CATEGORY_SHUFFLE) |
| .doc( |
| "Whether to enable round robin partitioning for Comet native shuffle. " + |
| "This is disabled by default because Comet's round-robin produces different " + |
| "partition assignments than Spark. Spark sorts rows by their binary UnsafeRow " + |
| "representation before assigning partitions, but Comet uses Arrow format which " + |
| "has a different binary layout. Instead, Comet implements round-robin as hash " + |
| "partitioning on all columns, which achieves the same goals: even distribution, " + |
| "deterministic output (for fault tolerance), and no semantic grouping. " + |
| "Sorted output will be identical to Spark, but unsorted row ordering may differ.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_EXEC_SHUFFLE_WITH_ROUND_ROBIN_PARTITIONING_MAX_HASH_COLUMNS: ConfigEntry[Int] = |
| conf("spark.comet.native.shuffle.partitioning.roundrobin.maxHashColumns") |
| .category(CATEGORY_SHUFFLE) |
| .doc( |
| "The maximum number of columns to hash for round robin partitioning. " + |
| "When set to 0 (the default), all columns are hashed. " + |
| "When set to a positive value, only the first N columns are used for hashing, " + |
| "which can improve performance for wide tables while still providing " + |
| "reasonable distribution.") |
| .intConf |
| .checkValue( |
| v => v >= 0, |
| "The maximum number of columns to hash for round robin partitioning must be non-negative.") |
| .createWithDefault(0) |
| |
| val COMET_EXEC_SHUFFLE_COMPRESSION_CODEC: ConfigEntry[String] = |
| conf(s"$COMET_EXEC_CONFIG_PREFIX.shuffle.compression.codec") |
| .category(CATEGORY_SHUFFLE) |
| .doc( |
| "The codec of Comet native shuffle used to compress shuffle data. lz4, zstd, and " + |
| "snappy are supported. Compression can be disabled by setting " + |
| "spark.shuffle.compress=false.") |
| .stringConf |
| .checkValues(Set("zstd", "lz4", "snappy")) |
| .createWithDefault("lz4") |
| |
| val COMET_EXEC_SHUFFLE_COMPRESSION_ZSTD_LEVEL: ConfigEntry[Int] = |
| conf(s"$COMET_EXEC_CONFIG_PREFIX.shuffle.compression.zstd.level") |
| .category(CATEGORY_SHUFFLE) |
| .doc("The compression level to use when compressing shuffle files with zstd.") |
| .intConf |
| .createWithDefault(1) |
| |
| val COMET_COLUMNAR_SHUFFLE_ASYNC_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.columnar.shuffle.async.enabled") |
| .category(CATEGORY_SHUFFLE) |
| .doc("Whether to enable asynchronous shuffle for Arrow-based shuffle.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_COLUMNAR_SHUFFLE_ASYNC_THREAD_NUM: ConfigEntry[Int] = |
| conf("spark.comet.columnar.shuffle.async.thread.num") |
| .category(CATEGORY_SHUFFLE) |
| .doc( |
| "Number of threads used for Comet async columnar shuffle per shuffle task. " + |
| "Note that more threads means more memory requirement to " + |
| "buffer shuffle data before flushing to disk. Also, more threads may not always " + |
| "improve performance, and should be set based on the number of cores available.") |
| .intConf |
| .createWithDefault(3) |
| |
| val COMET_COLUMNAR_SHUFFLE_ASYNC_MAX_THREAD_NUM: ConfigEntry[Int] = { |
| conf("spark.comet.columnar.shuffle.async.max.thread.num") |
| .category(CATEGORY_SHUFFLE) |
| .doc("Maximum number of threads on an executor used for Comet async columnar shuffle. " + |
| "This is the upper bound of total number of shuffle " + |
| "threads per executor. In other words, if the number of cores * the number of shuffle " + |
| "threads per task `spark.comet.columnar.shuffle.async.thread.num` is larger than " + |
| "this config. Comet will use this config as the number of shuffle threads per " + |
| "executor instead.") |
| .intConf |
| .createWithDefault(100) |
| } |
| |
| val COMET_COLUMNAR_SHUFFLE_SPILL_THRESHOLD: ConfigEntry[Int] = |
| conf("spark.comet.columnar.shuffle.spill.threshold") |
| .category(CATEGORY_SHUFFLE) |
| .doc( |
| "Number of rows to be spilled used for Comet columnar shuffle. " + |
| "For every configured number of rows, a new spill file will be created. " + |
| "Higher value means more memory requirement to buffer shuffle data before " + |
| "flushing to disk. As Comet uses columnar shuffle which is columnar format, " + |
| "higher value usually helps to improve shuffle data compression ratio. This is " + |
| "internal config for testing purpose or advanced tuning.") |
| .internal() |
| .intConf |
| .createWithDefault(Int.MaxValue) |
| |
| val COMET_ONHEAP_SHUFFLE_MEMORY_FACTOR: ConfigEntry[Double] = |
| conf("spark.comet.columnar.shuffle.memory.factor") |
| .category(CATEGORY_TESTING) |
| .doc("Fraction of Comet memory to be allocated per executor process for columnar shuffle " + |
| s"when running in on-heap mode. $TUNING_GUIDE.") |
| .doubleConf |
| .checkValue( |
| factor => factor > 0, |
| "Ensure that Comet shuffle memory overhead factor is a double greater than 0") |
| .createWithDefault(1.0) |
| |
| val COMET_BATCH_SIZE: ConfigEntry[Int] = conf("spark.comet.batchSize") |
| .category(CATEGORY_TUNING) |
| .doc("The columnar batch size, i.e., the maximum number of rows that a batch can contain.") |
| .intConf |
| .checkValue(v => v > 0, "Batch size must be positive") |
| .createWithDefault(8192) |
| |
| val COMET_COLUMNAR_SHUFFLE_BATCH_SIZE: ConfigEntry[Int] = |
| conf("spark.comet.columnar.shuffle.batch.size") |
| .category(CATEGORY_SHUFFLE) |
| .doc("Batch size when writing out sorted spill files on the native side. Note that " + |
| "this should not be larger than batch size (i.e., `spark.comet.batchSize`). Otherwise " + |
| "it will produce larger batches than expected in the native operator after shuffle.") |
| .intConf |
| .checkValue( |
| v => v <= COMET_BATCH_SIZE.get(), |
| "Should not be larger than batch size `spark.comet.batchSize`") |
| .createWithDefault(8192) |
| |
| val COMET_SHUFFLE_WRITE_BUFFER_SIZE: ConfigEntry[Long] = |
| conf(s"$COMET_EXEC_CONFIG_PREFIX.shuffle.writeBufferSize") |
| .category(CATEGORY_SHUFFLE) |
| .doc("Size of the write buffer in bytes used by the native shuffle writer when writing " + |
| "shuffle data to disk. Larger values may improve write performance by reducing " + |
| "the number of system calls, but will use more memory. " + |
| "The default is 1MB which provides a good balance between performance and memory usage.") |
| .bytesConf(ByteUnit.MiB) |
| .checkValue(v => v > 0, "Write buffer size must be positive") |
| .createWithDefault(1) |
| |
| val COMET_SHUFFLE_PREFER_DICTIONARY_RATIO: ConfigEntry[Double] = conf( |
| "spark.comet.shuffle.preferDictionary.ratio") |
| .category(CATEGORY_SHUFFLE) |
| .doc( |
| "The ratio of total values to distinct values in a string column to decide whether to " + |
| "prefer dictionary encoding when shuffling the column. If the ratio is higher than " + |
| "this config, dictionary encoding will be used on shuffling string column. This config " + |
| "is effective if it is higher than 1.0. Note that this " + |
| "config is only used when `spark.comet.exec.shuffle.mode` is `jvm`.") |
| .doubleConf |
| .createWithDefault(10.0) |
| |
| val COMET_EXCHANGE_SIZE_MULTIPLIER: ConfigEntry[Double] = conf( |
| "spark.comet.shuffle.sizeInBytesMultiplier") |
| .category(CATEGORY_SHUFFLE) |
| .doc( |
| "Comet reports smaller sizes for shuffle due to using Arrow's columnar memory format " + |
| "and this can result in Spark choosing a different join strategy due to the estimated " + |
| "size of the exchange being smaller. Comet will multiple sizeInBytes by this amount to " + |
| "avoid regressions in join strategy.") |
| .doubleConf |
| .createWithDefault(1.0) |
| |
| val COMET_DPP_FALLBACK_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.dppFallback.enabled") |
| .category(CATEGORY_EXEC) |
| .doc("Whether to fall back to Spark for queries that use DPP.") |
| .booleanConf |
| .createWithDefault(true) |
| |
| val COMET_DEBUG_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.debug.enabled") |
| .category(CATEGORY_EXEC) |
| .doc( |
| "Whether to enable debug mode for Comet. " + |
| "When enabled, Comet will do additional checks for debugging purpose. For example, " + |
| "validating array when importing arrays from JVM at native side. Note that these " + |
| "checks may be expensive in performance and should only be enabled for debugging " + |
| "purpose.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| // Used on native side. Check spark_config.rs how the config is used |
| val COMET_DEBUG_MEMORY_ENABLED: ConfigEntry[Boolean] = |
| conf(s"$COMET_PREFIX.debug.memory") |
| .category(CATEGORY_TESTING) |
| .doc(s"When enabled, log all native memory pool interactions. $DEBUGGING_GUIDE.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_EXTENDED_EXPLAIN_FORMAT_VERBOSE = "verbose" |
| val COMET_EXTENDED_EXPLAIN_FORMAT_FALLBACK = "fallback" |
| |
| val COMET_EXTENDED_EXPLAIN_FORMAT: ConfigEntry[String] = |
| conf("spark.comet.explain.format") |
| .category(CATEGORY_EXEC_EXPLAIN) |
| .doc("Choose extended explain output. The default format of " + |
| s"'$COMET_EXTENDED_EXPLAIN_FORMAT_VERBOSE' will provide the full query plan annotated " + |
| "with fallback reasons as well as a summary of how much of the plan was accelerated " + |
| s"by Comet. The format '$COMET_EXTENDED_EXPLAIN_FORMAT_FALLBACK' provides a list of " + |
| "fallback reasons instead.") |
| .stringConf |
| .checkValues( |
| Set(COMET_EXTENDED_EXPLAIN_FORMAT_VERBOSE, COMET_EXTENDED_EXPLAIN_FORMAT_FALLBACK)) |
| .createWithDefault(COMET_EXTENDED_EXPLAIN_FORMAT_VERBOSE) |
| |
| val COMET_EXPLAIN_NATIVE_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.explain.native.enabled") |
| .category(CATEGORY_EXEC_EXPLAIN) |
| .doc( |
| "When this setting is enabled, Comet will provide a tree representation of " + |
| "the native query plan before execution and again after execution, with " + |
| "metrics.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_EXPLAIN_TRANSFORMATIONS: ConfigEntry[Boolean] = |
| conf("spark.comet.explain.rules") |
| .category(CATEGORY_EXEC_EXPLAIN) |
| .doc("When this setting is enabled, Comet will log all plan transformations performed " + |
| "in physical optimizer rules. Default: false") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_LOG_FALLBACK_REASONS: ConfigEntry[Boolean] = |
| conf("spark.comet.logFallbackReasons.enabled") |
| .category(CATEGORY_EXEC_EXPLAIN) |
| .doc("When this setting is enabled, Comet will log warnings for all fallback reasons.") |
| .booleanConf |
| .createWithEnvVarOrDefault("ENABLE_COMET_LOG_FALLBACK_REASONS", false) |
| |
| val COMET_EXPLAIN_FALLBACK_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.explainFallback.enabled") |
| .category(CATEGORY_EXEC_EXPLAIN) |
| .doc( |
| "When this setting is enabled, Comet will provide logging explaining the reason(s) " + |
| "why a query stage cannot be executed natively. Set this to false to " + |
| "reduce the amount of logging.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_PARQUET_ENABLE_DIRECT_BUFFER: ConfigEntry[Boolean] = |
| conf("spark.comet.parquet.enable.directBuffer") |
| .category(CATEGORY_PARQUET) |
| .doc("Whether to use Java direct byte buffer when reading Parquet.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_ONHEAP_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.exec.onHeap.enabled") |
| .category(CATEGORY_TESTING) |
| .doc("Whether to allow Comet to run in on-heap mode. Required for running Spark SQL tests.") |
| .booleanConf |
| .createWithEnvVarOrDefault("ENABLE_COMET_ONHEAP", false) |
| |
| val COMET_OFFHEAP_MEMORY_POOL_TYPE: ConfigEntry[String] = |
| conf("spark.comet.exec.memoryPool") |
| .category(CATEGORY_TUNING) |
| .doc( |
| "The type of memory pool to be used for Comet native execution when running Spark in " + |
| "off-heap mode. Available pool types are `greedy_unified` and `fair_unified`. " + |
| s"$TUNING_GUIDE.") |
| .stringConf |
| .createWithDefault("fair_unified") |
| |
| val COMET_ONHEAP_MEMORY_POOL_TYPE: ConfigEntry[String] = conf( |
| "spark.comet.exec.onHeap.memoryPool") |
| .category(CATEGORY_TESTING) |
| .doc( |
| "The type of memory pool to be used for Comet native execution " + |
| "when running Spark in on-heap mode. Available pool types are `greedy`, `fair_spill`, " + |
| "`greedy_task_shared`, `fair_spill_task_shared`, `greedy_global`, `fair_spill_global`, " + |
| "and `unbounded`.") |
| .stringConf |
| .createWithDefault("greedy_task_shared") |
| |
| val COMET_OFFHEAP_MEMORY_POOL_FRACTION: ConfigEntry[Double] = |
| conf("spark.comet.exec.memoryPool.fraction") |
| .category(CATEGORY_TUNING) |
| .doc( |
| "Fraction of off-heap memory pool that is available to Comet. " + |
| "Only applies to off-heap mode. " + |
| s"$TUNING_GUIDE.") |
| .doubleConf |
| .createWithDefault(1.0) |
| |
| val COMET_SCAN_PREFETCH_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.scan.preFetch.enabled") |
| .category(CATEGORY_SCAN) |
| .doc("Whether to enable pre-fetching feature of CometScan.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_SCAN_PREFETCH_THREAD_NUM: ConfigEntry[Int] = |
| conf("spark.comet.scan.preFetch.threadNum") |
| .category(CATEGORY_SCAN) |
| .doc( |
| "The number of threads running pre-fetching for CometScan. Effective if " + |
| s"${COMET_SCAN_PREFETCH_ENABLED.key} is enabled. Note that more " + |
| "pre-fetching threads means more memory requirement to store pre-fetched row groups.") |
| .intConf |
| .createWithDefault(2) |
| |
| val COMET_NATIVE_LOAD_REQUIRED: ConfigEntry[Boolean] = conf("spark.comet.nativeLoadRequired") |
| .category(CATEGORY_EXEC) |
| .doc( |
| "Whether to require Comet native library to load successfully when Comet is enabled. " + |
| "If not, Comet will silently fallback to Spark when it fails to load the native lib. " + |
| "Otherwise, an error will be thrown and the Spark job will be aborted.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_EXCEPTION_ON_LEGACY_DATE_TIMESTAMP: ConfigEntry[Boolean] = |
| conf("spark.comet.exceptionOnDatetimeRebase") |
| .category(CATEGORY_EXEC) |
| .doc("Whether to throw exception when seeing dates/timestamps from the legacy hybrid " + |
| "(Julian + Gregorian) calendar. Since Spark 3, dates/timestamps were written according " + |
| "to the Proleptic Gregorian calendar. When this is true, Comet will " + |
| "throw exceptions when seeing these dates/timestamps that were written by Spark version " + |
| "before 3.0. If this is false, these dates/timestamps will be read as if they were " + |
| "written to the Proleptic Gregorian calendar and will not be rebased.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_USE_DECIMAL_128: ConfigEntry[Boolean] = conf("spark.comet.use.decimal128") |
| .internal() |
| .category(CATEGORY_EXEC) |
| .doc("If true, Comet will always use 128 bits to represent a decimal value, regardless of " + |
| "its precision. If false, Comet will use 32, 64 and 128 bits respectively depending on " + |
| "the precision. N.B. this is NOT a user-facing config but should be inferred and set by " + |
| "Comet itself.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_USE_LAZY_MATERIALIZATION: ConfigEntry[Boolean] = conf( |
| "spark.comet.use.lazyMaterialization") |
| .internal() |
| .category(CATEGORY_PARQUET) |
| .doc( |
| "Whether to enable lazy materialization for Comet. When this is turned on, Comet will " + |
| "read Parquet data source lazily for string and binary columns. For filter operations, " + |
| "lazy materialization will improve read performance by skipping unused pages.") |
| .booleanConf |
| .createWithDefault(true) |
| |
| val COMET_SCHEMA_EVOLUTION_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.schemaEvolution.enabled") |
| .internal() |
| .category(CATEGORY_SCAN) |
| .doc("Whether to enable schema evolution in Comet. For instance, promoting a integer " + |
| "column to a long column, a float column to a double column, etc. This is automatically" + |
| "enabled when reading from Iceberg tables.") |
| .booleanConf |
| .createWithDefault(COMET_SCHEMA_EVOLUTION_ENABLED_DEFAULT) |
| |
| val COMET_ENABLE_PARTIAL_HASH_AGGREGATE: ConfigEntry[Boolean] = |
| conf("spark.comet.testing.aggregate.partialMode.enabled") |
| .internal() |
| .category(CATEGORY_TESTING) |
| .doc("This setting is used in unit tests") |
| .booleanConf |
| .createWithDefault(true) |
| |
| val COMET_ENABLE_FINAL_HASH_AGGREGATE: ConfigEntry[Boolean] = |
| conf("spark.comet.testing.aggregate.finalMode.enabled") |
| .internal() |
| .category(CATEGORY_TESTING) |
| .doc("This setting is used in unit tests") |
| .booleanConf |
| .createWithDefault(true) |
| |
| val COMET_SPARK_TO_ARROW_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.sparkToColumnar.enabled") |
| .category(CATEGORY_TESTING) |
| .doc("Whether to enable Spark to Arrow columnar conversion. When this is turned on, " + |
| "Comet will convert operators in " + |
| "`spark.comet.sparkToColumnar.supportedOperatorList` into Arrow columnar format before " + |
| "processing. This is an experimental feature and has known issues with non-UTC timezones.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST: ConfigEntry[Seq[String]] = |
| conf("spark.comet.sparkToColumnar.supportedOperatorList") |
| .category(CATEGORY_TESTING) |
| .doc("A comma-separated list of operators that will be converted to Arrow columnar " + |
| s"format when `${COMET_SPARK_TO_ARROW_ENABLED.key}` is true.") |
| .stringConf |
| .toSequence |
| .createWithDefault(Seq("Range,InMemoryTableScan,RDDScan")) |
| |
| val COMET_CASE_CONVERSION_ENABLED: ConfigEntry[Boolean] = |
| conf("spark.comet.caseConversion.enabled") |
| .category(CATEGORY_EXEC) |
| .doc("Java uses locale-specific rules when converting strings to upper or lower case and " + |
| "Rust does not, so we disable upper and lower by default.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_PARQUET_UNSIGNED_SMALL_INT_CHECK: ConfigEntry[Boolean] = |
| conf("spark.comet.scan.unsignedSmallIntSafetyCheck") |
| .category(CATEGORY_SCAN) |
| .doc( |
| "Parquet files may contain unsigned 8-bit integers (UINT_8) which Spark maps to " + |
| "ShortType. When this config is true (default), Comet falls back to Spark for " + |
| "ShortType columns because we cannot distinguish signed INT16 (safe) from unsigned " + |
| "UINT_8 (may produce different results). Set to false to allow native execution of " + |
| "ShortType columns if you know your data does not contain unsigned UINT_8 columns " + |
| s"from improperly encoded Parquet files. $COMPAT_GUIDE.") |
| .booleanConf |
| .createWithDefault(true) |
| |
| val COMET_EXEC_STRICT_FLOATING_POINT: ConfigEntry[Boolean] = |
| conf("spark.comet.exec.strictFloatingPoint") |
| .category(CATEGORY_EXEC) |
| .doc( |
| "When enabled, fall back to Spark for floating-point operations that may differ from " + |
| s"Spark, such as when comparing or sorting -0.0 and 0.0. $COMPAT_GUIDE.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_METRICS_UPDATE_INTERVAL: ConfigEntry[Long] = |
| conf("spark.comet.metrics.updateInterval") |
| .category(CATEGORY_EXEC) |
| .doc("The interval in milliseconds to update metrics. If interval is negative," + |
| " metrics will be updated upon task completion.") |
| .longConf |
| .createWithDefault(3000L) |
| |
| val COMET_LIBHDFS_SCHEMES_KEY = "fs.comet.libhdfs.schemes" |
| |
| val COMET_LIBHDFS_SCHEMES: OptionalConfigEntry[String] = |
| conf(s"spark.hadoop.$COMET_LIBHDFS_SCHEMES_KEY") |
| .category(CATEGORY_SCAN) |
| .doc("Defines filesystem schemes (e.g., hdfs, webhdfs) that the native side accesses " + |
| "via libhdfs, separated by commas. Valid only when built with hdfs feature enabled.") |
| .stringConf |
| .createOptional |
| |
| // Used on native side. Check spark_config.rs how the config is used |
| val COMET_MAX_TEMP_DIRECTORY_SIZE: ConfigEntry[Long] = |
| conf("spark.comet.maxTempDirectorySize") |
| .category(CATEGORY_EXEC) |
| .doc("The maximum amount of data (in bytes) stored inside the temporary directories.") |
| .bytesConf(ByteUnit.BYTE) |
| .createWithDefault(100L * 1024 * 1024 * 1024) // 100 GB |
| |
| val COMET_RESPECT_DATAFUSION_CONFIGS: ConfigEntry[Boolean] = |
| conf(s"$COMET_EXEC_CONFIG_PREFIX.respectDataFusionConfigs") |
| .category(CATEGORY_TESTING) |
| .doc( |
| "Development and testing configuration option to allow DataFusion configs set in " + |
| "Spark configuration settings starting with `spark.comet.datafusion.` to be passed " + |
| "into native execution.") |
| .booleanConf |
| .createWithDefault(false) |
| |
| val COMET_STRICT_TESTING: ConfigEntry[Boolean] = conf(s"$COMET_PREFIX.testing.strict") |
| .category(CATEGORY_TESTING) |
| .doc("Experimental option to enable strict testing, which will fail tests that could be " + |
| "more comprehensive, such as checking for a specific fallback reason.") |
| .booleanConf |
| .createWithEnvVarOrDefault("ENABLE_COMET_STRICT_TESTING", false) |
| |
| val COMET_OPERATOR_DATA_WRITING_COMMAND_ALLOW_INCOMPAT: ConfigEntry[Boolean] = |
| createOperatorIncompatConfig("DataWritingCommandExec") |
| |
| /** Create a config to enable a specific operator */ |
| private def createExecEnabledConfig( |
| exec: String, |
| defaultValue: Boolean, |
| notes: Option[String] = None): ConfigEntry[Boolean] = { |
| conf(s"$COMET_EXEC_CONFIG_PREFIX.$exec.enabled") |
| .category(CATEGORY_ENABLE_EXEC) |
| .doc( |
| s"Whether to enable $exec by default." + notes |
| .map(s => s" $s.") |
| .getOrElse("")) |
| .booleanConf |
| .createWithDefault(defaultValue) |
| } |
| |
| /** |
| * Converts a config key to a valid environment variable name. Example: |
| * "spark.comet.operator.DataWritingCommandExec.allowIncompatible" -> |
| * "SPARK_COMET_OPERATOR_DATAWRITINGCOMMANDEXEC_ALLOWINCOMPATIBLE" |
| */ |
| private def configKeyToEnvVar(configKey: String): String = |
| configKey.toUpperCase(Locale.ROOT).replace('.', '_') |
| |
| private def createOperatorIncompatConfig(name: String): ConfigEntry[Boolean] = { |
| val configKey = getOperatorAllowIncompatConfigKey(name) |
| val envVar = configKeyToEnvVar(configKey) |
| conf(configKey) |
| .category(CATEGORY_EXEC) |
| .doc(s"Whether to allow incompatibility for operator: $name. " + |
| s"False by default. Can be overridden with $envVar env variable") |
| .booleanConf |
| .createWithEnvVarOrDefault(envVar, false) |
| } |
| |
| def isExprEnabled(name: String, conf: SQLConf = SQLConf.get): Boolean = { |
| getBooleanConf(getExprEnabledConfigKey(name), defaultValue = true, conf) |
| } |
| |
| def getExprEnabledConfigKey(name: String): String = { |
| s"${CometConf.COMET_EXPR_CONFIG_PREFIX}.$name.enabled" |
| } |
| |
| def isExprAllowIncompat(name: String, conf: SQLConf = SQLConf.get): Boolean = { |
| getBooleanConf(getExprAllowIncompatConfigKey(name), defaultValue = false, conf) |
| } |
| |
| def getExprAllowIncompatConfigKey(name: String): String = { |
| s"${CometConf.COMET_EXPR_CONFIG_PREFIX}.$name.allowIncompatible" |
| } |
| |
| def getExprAllowIncompatConfigKey(exprClass: Class[_]): String = { |
| s"${CometConf.COMET_EXPR_CONFIG_PREFIX}.${exprClass.getSimpleName}.allowIncompatible" |
| } |
| |
| def isOperatorAllowIncompat(name: String, conf: SQLConf = SQLConf.get): Boolean = { |
| getBooleanConf(getOperatorAllowIncompatConfigKey(name), defaultValue = false, conf) |
| } |
| |
| def getOperatorAllowIncompatConfigKey(name: String): String = { |
| s"${CometConf.COMET_OPERATOR_CONFIG_PREFIX}.$name.allowIncompatible" |
| } |
| |
| def getOperatorAllowIncompatConfigKey(exprClass: Class[_]): String = { |
| s"${CometConf.COMET_OPERATOR_CONFIG_PREFIX}.${exprClass.getSimpleName}.allowIncompatible" |
| } |
| |
| def getBooleanConf(name: String, defaultValue: Boolean, conf: SQLConf): Boolean = { |
| conf.getConfString(name, defaultValue.toString).toLowerCase(Locale.ROOT) == "true" |
| } |
| } |
| |
| object ConfigHelpers { |
| def toNumber[T](s: String, converter: String => T, key: String, configType: String): T = { |
| try { |
| converter(s.trim) |
| } catch { |
| case _: NumberFormatException => |
| throw new IllegalArgumentException(s"$key should be $configType, but was $s") |
| } |
| } |
| |
| def toBoolean(s: String, key: String): Boolean = { |
| try { |
| s.trim.toBoolean |
| } catch { |
| case _: IllegalArgumentException => |
| throw new IllegalArgumentException(s"$key should be boolean, but was $s") |
| } |
| } |
| |
| def stringToSeq[T](str: String, converter: String => T): Seq[T] = { |
| Utils.stringToSeq(str).map(converter) |
| } |
| |
| def seqToString[T](v: Seq[T], stringConverter: T => String): String = { |
| v.map(stringConverter).mkString(",") |
| } |
| |
| def timeFromString(str: String, unit: TimeUnit): Long = JavaUtils.timeStringAs(str, unit) |
| |
| def timeToString(v: Long, unit: TimeUnit): String = |
| TimeUnit.MILLISECONDS.convert(v, unit) + "ms" |
| |
| def byteFromString(str: String, unit: ByteUnit): Long = { |
| val (input, multiplier) = |
| if (str.nonEmpty && str.charAt(0) == '-') { |
| (str.substring(1), -1) |
| } else { |
| (str, 1) |
| } |
| multiplier * JavaUtils.byteStringAs(input, unit) |
| } |
| |
| def byteToString(v: Long, unit: ByteUnit): String = unit.convertTo(v, ByteUnit.BYTE) + "b" |
| } |
| |
| private class TypedConfigBuilder[T]( |
| val parent: ConfigBuilder, |
| val converter: String => T, |
| val stringConverter: T => String) { |
| |
| import ConfigHelpers._ |
| |
| def this(parent: ConfigBuilder, converter: String => T) = { |
| this(parent, converter, Option(_).map(_.toString).orNull) |
| } |
| |
| /** Apply a transformation to the user-provided values of the config entry. */ |
| def transform(fn: T => T): TypedConfigBuilder[T] = { |
| new TypedConfigBuilder(parent, s => fn(converter(s)), stringConverter) |
| } |
| |
| /** Checks if the user-provided value for the config matches the validator. */ |
| def checkValue(validator: T => Boolean, errorMsg: String): TypedConfigBuilder[T] = { |
| transform { v => |
| if (!validator(v)) { |
| throw new IllegalArgumentException(s"'$v' in ${parent.key} is invalid. $errorMsg") |
| } |
| v |
| } |
| } |
| |
| /** Check that user-provided values for the config match a pre-defined set. */ |
| def checkValues(validValues: Set[T]): TypedConfigBuilder[T] = { |
| transform { v => |
| if (!validValues.contains(v)) { |
| throw new IllegalArgumentException( |
| s"The value of ${parent.key} should be one of ${validValues.mkString(", ")}, but was $v") |
| } |
| v |
| } |
| } |
| |
| /** Turns the config entry into a sequence of values of the underlying type. */ |
| def toSequence: TypedConfigBuilder[Seq[T]] = { |
| new TypedConfigBuilder(parent, stringToSeq(_, converter), seqToString(_, stringConverter)) |
| } |
| |
| /** Creates a [[ConfigEntry]] that does not have a default value. */ |
| def createOptional: OptionalConfigEntry[T] = { |
| val conf = new OptionalConfigEntry[T]( |
| parent.key, |
| converter, |
| stringConverter, |
| parent._doc, |
| parent._category, |
| parent._public, |
| parent._version) |
| CometConf.register(conf) |
| conf |
| } |
| |
| /** Creates a [[ConfigEntry]] that has a default value. */ |
| def createWithDefault(default: T): ConfigEntry[T] = { |
| val transformedDefault = converter(stringConverter(default)) |
| val conf = new ConfigEntryWithDefault[T]( |
| parent.key, |
| transformedDefault, |
| converter, |
| stringConverter, |
| parent._doc, |
| parent._category, |
| parent._public, |
| parent._version) |
| CometConf.register(conf) |
| conf |
| } |
| |
| /** |
| * Creates a [[ConfigEntry]] that has a default value, with support for environment variable |
| * override. |
| * |
| * The value is resolved in the following priority order: |
| * 1. Spark config value (if set) 2. Environment variable value (if set) 3. Default value |
| * |
| * @param envVar |
| * The environment variable name to check for override value |
| * @param default |
| * The default value to use if neither config nor env var is set |
| * @return |
| * A ConfigEntry with environment variable support |
| */ |
| def createWithEnvVarOrDefault(envVar: String, default: T): ConfigEntry[T] = { |
| val transformedDefault = converter(sys.env.getOrElse(envVar, stringConverter(default))) |
| val conf = new ConfigEntryWithDefault[T]( |
| parent.key, |
| transformedDefault, |
| converter, |
| stringConverter, |
| parent._doc, |
| parent._category, |
| parent._public, |
| parent._version, |
| Some(envVar)) |
| CometConf.register(conf) |
| conf |
| } |
| } |
| |
| abstract class ConfigEntry[T]( |
| val key: String, |
| val valueConverter: String => T, |
| val stringConverter: T => String, |
| val doc: String, |
| val category: String, |
| val isPublic: Boolean, |
| val version: String) { |
| |
| /** |
| * Retrieves the config value from the given [[SQLConf]]. |
| */ |
| def get(conf: SQLConf): T |
| |
| /** |
| * Retrieves the config value from the current thread-local [[SQLConf]] |
| * |
| * @return |
| */ |
| def get(): T = get(SQLConf.get) |
| |
| def defaultValue: Option[T] = None |
| |
| def defaultValueString: String |
| |
| /** |
| * The environment variable name that can override this config's default value, if applicable. |
| */ |
| def envVar: Option[String] = None |
| |
| override def toString: String = { |
| s"ConfigEntry(key=$key, defaultValue=$defaultValueString, doc=$doc, " + |
| s"public=$isPublic, version=$version)" |
| } |
| } |
| |
| private[comet] class ConfigEntryWithDefault[T]( |
| key: String, |
| _defaultValue: T, |
| valueConverter: String => T, |
| stringConverter: T => String, |
| doc: String, |
| category: String, |
| isPublic: Boolean, |
| version: String, |
| _envVar: Option[String] = None) |
| extends ConfigEntry(key, valueConverter, stringConverter, doc, category, isPublic, version) { |
| override def defaultValue: Option[T] = Some(_defaultValue) |
| |
| override def defaultValueString: String = stringConverter(_defaultValue) |
| |
| override def envVar: Option[String] = _envVar |
| |
| def get(conf: SQLConf): T = { |
| val tmp = conf.getConfString(key, null) |
| if (tmp == null) { |
| _defaultValue |
| } else { |
| valueConverter(tmp) |
| } |
| } |
| } |
| |
| private[comet] class OptionalConfigEntry[T]( |
| key: String, |
| val rawValueConverter: String => T, |
| val rawStringConverter: T => String, |
| doc: String, |
| category: String, |
| isPublic: Boolean, |
| version: String) |
| extends ConfigEntry[Option[T]]( |
| key, |
| s => Some(rawValueConverter(s)), |
| v => v.map(rawStringConverter).orNull, |
| doc, |
| category, |
| isPublic, |
| version) { |
| |
| override def defaultValueString: String = ConfigEntry.UNDEFINED |
| |
| override def get(conf: SQLConf): Option[T] = { |
| Option(conf.getConfString(key, null)).map(rawValueConverter) |
| } |
| } |
| |
| private[comet] case class ConfigBuilder(key: String) { |
| |
| import ConfigHelpers._ |
| |
| var _public = true |
| var _doc = "" |
| var _version = "" |
| var _category = "" |
| |
| def internal(): ConfigBuilder = { |
| _public = false |
| this |
| } |
| |
| def doc(s: String): ConfigBuilder = { |
| _doc = s |
| this |
| } |
| |
| def category(s: String): ConfigBuilder = { |
| _category = s |
| this |
| } |
| |
| def version(v: String): ConfigBuilder = { |
| _version = v |
| this |
| } |
| |
| def intConf: TypedConfigBuilder[Int] = { |
| new TypedConfigBuilder(this, toNumber(_, _.toInt, key, "int")) |
| } |
| |
| def longConf: TypedConfigBuilder[Long] = { |
| new TypedConfigBuilder(this, toNumber(_, _.toLong, key, "long")) |
| } |
| |
| def doubleConf: TypedConfigBuilder[Double] = { |
| new TypedConfigBuilder(this, toNumber(_, _.toDouble, key, "double")) |
| } |
| |
| def booleanConf: TypedConfigBuilder[Boolean] = { |
| new TypedConfigBuilder(this, toBoolean(_, key)) |
| } |
| |
| def stringConf: TypedConfigBuilder[String] = { |
| new TypedConfigBuilder(this, v => v) |
| } |
| |
| def timeConf(unit: TimeUnit): TypedConfigBuilder[Long] = { |
| new TypedConfigBuilder(this, timeFromString(_, unit), timeToString(_, unit)) |
| } |
| |
| def bytesConf(unit: ByteUnit): TypedConfigBuilder[Long] = { |
| new TypedConfigBuilder(this, byteFromString(_, unit), byteToString(_, unit)) |
| } |
| } |
| |
| private object ConfigEntry { |
| val UNDEFINED = "<undefined>" |
| } |
