Skip to content

Spark: Support read with settings #367

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Dec 9, 2024
Merged

Spark: Support read with settings #367

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ccd6d3f
allow read with settings
huashi-st Nov 8, 2024
1a7ea06
make optional
huashi-st Nov 9, 2024
db3cd5a
update doc
huashi-st Nov 9, 2024
ddd62fd
update doc to describe default value is None
huashi-st Dec 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/configurations/02_sql_configurations.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ spark.clickhouse.ignoreUnsupportedTransform|false|ClickHouse supports using comp
spark.clickhouse.read.compression.codec|lz4|The codec used to decompress data for reading. Supported codecs: none, lz4.|0.5.0
spark.clickhouse.read.distributed.convertLocal|true|When reading Distributed table, read local table instead of itself. If `true`, ignore `spark.clickhouse.read.distributed.useClusterNodes`.|0.1.0
spark.clickhouse.read.fixedStringAs|binary|Read ClickHouse FixedString type as the specified Spark data type. Supported types: binary, string|0.8.0
spark.clickhouse.read.withSettings|""|Query-level settings when reading from ClickHouse. e.g. `final=1, max_execution_time=5`|0.9.0
spark.clickhouse.read.format|json|Serialize format for reading. Supported formats: json, binary|0.6.0
spark.clickhouse.read.runtimeFilter.enabled|false|Enable runtime filter for reading.|0.8.0
spark.clickhouse.read.splitByPartitionId|true|If `true`, construct input partition filter by virtual column `_partition_id`, instead of partition value. There are known bugs to assemble SQL predication by partition value. This feature requires ClickHouse Server v21.6+|0.4.0
Expand Down
2 changes: 2 additions & 0 deletions spark-3.3/clickhouse-spark/src/main/scala/com/clickhouse/spark/read/ClickHouseReader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ abstract class ClickHouseReader[Record](

val readDistributedUseClusterNodes: Boolean = conf.getConf(READ_DISTRIBUTED_USE_CLUSTER_NODES)
val readDistributedConvertLocal: Boolean = conf.getConf(READ_DISTRIBUTED_CONVERT_LOCAL)
private val readWithSettings: String = conf.getConf(READ_WITH_SETTINGS)

val database: String = part.table.database
val table: String = part.table.name
Expand All @@ -60,6 +61,7 @@ abstract class ClickHouseReader[Record](
|WHERE (${part.partFilterExpr}) AND (${scanJob.filtersExpr})
|${scanJob.groupByClause.getOrElse("")}
|${scanJob.limit.map(n => s"LIMIT $n").getOrElse("")}
|${if (readWithSettings.nonEmpty) s"SETTINGS $readWithSettings" else ""}
Copy link
Collaborator

@pan3793 pan3793 Nov 8, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
|${if (readWithSettings.nonEmpty) s"SETTINGS $readWithSettings" else ""}
|${readSettings.map(settings => s"SETTINGS $settings" else ""}

Copy link
Contributor Author

@harryshi10 harryshi10 Nov 9, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

updated

|""".stripMargin
}

Expand Down
9 changes: 9 additions & 0 deletions ...3/clickhouse-spark/src/main/scala/org/apache/spark/sql/clickhouse/ClickHouseSQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,4 +209,13 @@ object ClickHouseSQLConf {
.stringConf
.transform(_.toLowerCase)
.createWithDefault("binary")

val READ_WITH_SETTINGS: ConfigEntry[String] =
buildConf("spark.clickhouse.read.withSettings")
Copy link
Collaborator

@pan3793 pan3793 Nov 8, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
buildConf("spark.clickhouse.read.withSettings")
buildConf("spark.clickhouse.read.settings")

Copy link
Contributor Author

@harryshi10 harryshi10 Nov 8, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

will also renaming the val to READ_SETTINGS

Copy link
Contributor Author

@harryshi10 harryshi10 Nov 9, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

updated

.doc("Query-level settings when reading from ClickHouse. e.g. `final=1, max_execution_time=5`")
Copy link
Collaborator

@pan3793 pan3793 Nov 8, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

from the spark perspective, it's a session-level configuration

SET spark.clickhouse.read.settings=final=1;
SELECT * FROM ch.db.t1; -- affected
SELECT * FROM ch.db.t2; -- affected too

Copy link
Contributor Author

@harryshi10 harryshi10 Nov 8, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

agree. will update the doc

Copy link
Contributor Author

@harryshi10 harryshi10 Nov 9, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

updated

.version("0.9.0")
.stringConf
.transform(_.toLowerCase)
Copy link
Collaborator

@pan3793 pan3793 Nov 11, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why should convert the input to lower case?

Copy link
Contributor Author

@harryshi10 harryshi10 Nov 11, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

in case we'd like to log scanQuery which keeps SQL keywords in uppercase and other text in lowercase

.createWithDefault("")
Copy link
Collaborator

@pan3793 pan3793 Nov 8, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
.createWithDefault("")
.createOptional

Copy link
Contributor Author

@harryshi10 harryshi10 Nov 9, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

will update it to OptionalConfigEntry accordingly

Copy link
Contributor Author

@harryshi10 harryshi10 Nov 9, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

updated


}
2 changes: 2 additions & 0 deletions spark-3.4/clickhouse-spark/src/main/scala/com/clickhouse/spark/read/ClickHouseReader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ abstract class ClickHouseReader[Record](

val readDistributedUseClusterNodes: Boolean = conf.getConf(READ_DISTRIBUTED_USE_CLUSTER_NODES)
val readDistributedConvertLocal: Boolean = conf.getConf(READ_DISTRIBUTED_CONVERT_LOCAL)
private val readWithSettings: String = conf.getConf(READ_WITH_SETTINGS)

val database: String = part.table.database
val table: String = part.table.name
Expand All @@ -60,6 +61,7 @@ abstract class ClickHouseReader[Record](
|WHERE (${part.partFilterExpr}) AND (${scanJob.filtersExpr})
|${scanJob.groupByClause.getOrElse("")}
|${scanJob.limit.map(n => s"LIMIT $n").getOrElse("")}
|${if (readWithSettings.nonEmpty) s"SETTINGS $readWithSettings" else ""}
|""".stripMargin
}

Expand Down
9 changes: 9 additions & 0 deletions ...4/clickhouse-spark/src/main/scala/org/apache/spark/sql/clickhouse/ClickHouseSQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,4 +209,13 @@ object ClickHouseSQLConf {
.stringConf
.transform(_.toLowerCase)
.createWithDefault("binary")

val READ_WITH_SETTINGS: ConfigEntry[String] =
buildConf("spark.clickhouse.read.withSettings")
.doc("Query-level settings when reading from ClickHouse. e.g. `final=1, max_execution_time=5`")
.version("0.9.0")
.stringConf
.transform(_.toLowerCase)
.createWithDefault("")

}
2 changes: 2 additions & 0 deletions spark-3.5/clickhouse-spark/src/main/scala/com/clickhouse/spark/read/ClickHouseReader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ abstract class ClickHouseReader[Record](

val readDistributedUseClusterNodes: Boolean = conf.getConf(READ_DISTRIBUTED_USE_CLUSTER_NODES)
val readDistributedConvertLocal: Boolean = conf.getConf(READ_DISTRIBUTED_CONVERT_LOCAL)
private val readWithSettings: String = conf.getConf(READ_WITH_SETTINGS)

val database: String = part.table.database
val table: String = part.table.name
Expand All @@ -60,6 +61,7 @@ abstract class ClickHouseReader[Record](
|WHERE (${part.partFilterExpr}) AND (${scanJob.filtersExpr})
|${scanJob.groupByClause.getOrElse("")}
|${scanJob.limit.map(n => s"LIMIT $n").getOrElse("")}
|${if (readWithSettings.nonEmpty) s"SETTINGS $readWithSettings" else ""}
|""".stripMargin
}

Expand Down
9 changes: 9 additions & 0 deletions ...5/clickhouse-spark/src/main/scala/org/apache/spark/sql/clickhouse/ClickHouseSQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,4 +209,13 @@ object ClickHouseSQLConf {
.stringConf
.transform(_.toLowerCase)
.createWithDefault("binary")

val READ_WITH_SETTINGS: ConfigEntry[String] =
buildConf("spark.clickhouse.read.withSettings")
.doc("Query-level settings when reading from ClickHouse. e.g. `final=1, max_execution_time=5`")
.version("0.9.0")
.stringConf
.transform(_.toLowerCase)
.createWithDefault("")

}
Loading