You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by li...@apache.org on 2024/06/24 05:55:42 UTC
(incubator-gluten) branch main updated: [GLUTEN-6178][CH] Add config to insert remote file system directly #6192
This is an automated email from the ASF dual-hosted git repository.
liuneng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-gluten.git
The following commit(s) were added to refs/heads/main by this push:
new e0fcfe586 [GLUTEN-6178][CH] Add config to insert remote file system directly #6192
e0fcfe586 is described below
commit e0fcfe586efc7efb3ec0c349d5ca8b2371d969d4
Author: Shuai li <lo...@live.cn>
AuthorDate: Mon Jun 24 13:55:39 2024 +0800
[GLUTEN-6178][CH] Add config to insert remote file system directly #6192
What changes were proposed in this pull request?
(Please fill in changes proposed in this fix)
(Fixes: #6178)
How was this patch tested?
Test by ut
---
...GlutenClickHouseMergeTreeWriteOnHDFSSuite.scala | 44 +++++++++++++++++++++-
cpp-ch/local-engine/Common/CHUtil.cpp | 3 +-
cpp-ch/local-engine/Common/CHUtil.h | 4 +-
.../Disks/ObjectStorages/GlutenDiskHDFS.cpp | 10 ++++-
.../Disks/ObjectStorages/GlutenDiskHDFS.h | 2 +
.../Storages/Mergetree/SparkMergeTreeWriter.cpp | 40 +++++++++++---------
.../Storages/Mergetree/SparkMergeTreeWriter.h | 2 +
7 files changed, 83 insertions(+), 22 deletions(-)
diff --git a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseMergeTreeWriteOnHDFSSuite.scala b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseMergeTreeWriteOnHDFSSuite.scala
index 572d0cd50..99b212059 100644
--- a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseMergeTreeWriteOnHDFSSuite.scala
+++ b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseMergeTreeWriteOnHDFSSuite.scala
@@ -25,10 +25,12 @@ import org.apache.spark.sql.execution.datasources.v2.clickhouse.metadata.AddMerg
import org.apache.commons.io.FileUtils
import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.FileSystem
+import org.apache.hadoop.fs.{FileSystem, Path}
import java.io.File
+import scala.concurrent.duration.DurationInt
+
// Some sqls' line length exceeds 100
// scalastyle:off line.size.limit
@@ -614,5 +616,45 @@ class GlutenClickHouseMergeTreeWriteOnHDFSSuite
.count()
assertResult(600572)(result)
}
+
+ test("test mergetree insert with optimize basic") {
+ val tableName = "lineitem_mergetree_insert_optimize_basic_hdfs"
+ val dataPath = s"$HDFS_URL/test/$tableName"
+
+ withSQLConf(
+ "spark.databricks.delta.optimize.minFileSize" -> "200000000",
+ "spark.gluten.sql.columnar.backend.ch.runtime_settings.mergetree.merge_after_insert" -> "true",
+ "spark.gluten.sql.columnar.backend.ch.runtime_settings.mergetree.insert_without_local_storage" -> "true",
+ "spark.gluten.sql.columnar.backend.ch.runtime_settings.min_insert_block_size_rows" -> "10000"
+ ) {
+ spark.sql(s"""
+ |DROP TABLE IF EXISTS $tableName;
+ |""".stripMargin)
+
+ spark.sql(s"""
+ |CREATE TABLE IF NOT EXISTS $tableName
+ |USING clickhouse
+ |LOCATION '$dataPath'
+ |TBLPROPERTIES (storage_policy='__hdfs_main')
+ | as select * from lineitem
+ |""".stripMargin)
+
+ val ret = spark.sql(s"select count(*) from $tableName").collect()
+ assertResult(600572)(ret.apply(0).get(0))
+ val conf = new Configuration
+ conf.set("fs.defaultFS", HDFS_URL)
+ val fs = FileSystem.get(conf)
+
+ eventually(timeout(60.seconds), interval(2.seconds)) {
+ val it = fs.listFiles(new Path(dataPath), true)
+ var files = 0
+ while (it.hasNext) {
+ it.next()
+ files += 1
+ }
+ assertResult(72)(files)
+ }
+ }
+ }
}
// scalastyle:off line.size.limit
diff --git a/cpp-ch/local-engine/Common/CHUtil.cpp b/cpp-ch/local-engine/Common/CHUtil.cpp
index be66d8ecc..94cd38003 100644
--- a/cpp-ch/local-engine/Common/CHUtil.cpp
+++ b/cpp-ch/local-engine/Common/CHUtil.cpp
@@ -623,7 +623,8 @@ void BackendInitializerUtil::initSettings(std::map<std::string, std::string> & b
{
/// Initialize default setting.
settings.set("date_time_input_format", "best_effort");
- settings.set("mergetree.merge_after_insert", true);
+ settings.set(MERGETREE_MERGE_AFTER_INSERT, true);
+ settings.set(MERGETREE_INSERT_WITHOUT_LOCAL_STORAGE, false);
for (const auto & [key, value] : backend_conf_map)
{
diff --git a/cpp-ch/local-engine/Common/CHUtil.h b/cpp-ch/local-engine/Common/CHUtil.h
index 50de9461f..94e0f0168 100644
--- a/cpp-ch/local-engine/Common/CHUtil.h
+++ b/cpp-ch/local-engine/Common/CHUtil.h
@@ -35,7 +35,9 @@ class QueryPlan;
namespace local_engine
{
-static const std::unordered_set<String> BOOL_VALUE_SETTINGS{"mergetree.merge_after_insert"};
+static const String MERGETREE_INSERT_WITHOUT_LOCAL_STORAGE = "mergetree.insert_without_local_storage";
+static const String MERGETREE_MERGE_AFTER_INSERT = "mergetree.merge_after_insert";
+static const std::unordered_set<String> BOOL_VALUE_SETTINGS{MERGETREE_MERGE_AFTER_INSERT, MERGETREE_INSERT_WITHOUT_LOCAL_STORAGE};
static const std::unordered_set<String> LONG_VALUE_SETTINGS{
"optimize.maxfilesize", "optimize.minFileSize", "mergetree.max_num_part_per_merge_task"};
diff --git a/cpp-ch/local-engine/Disks/ObjectStorages/GlutenDiskHDFS.cpp b/cpp-ch/local-engine/Disks/ObjectStorages/GlutenDiskHDFS.cpp
index 07a7aa6bd..f207ad232 100644
--- a/cpp-ch/local-engine/Disks/ObjectStorages/GlutenDiskHDFS.cpp
+++ b/cpp-ch/local-engine/Disks/ObjectStorages/GlutenDiskHDFS.cpp
@@ -52,7 +52,15 @@ void GlutenDiskHDFS::createDirectories(const String & path)
void GlutenDiskHDFS::removeDirectory(const String & path)
{
DiskObjectStorage::removeDirectory(path);
- hdfsDelete(hdfs_object_storage->getHDFSFS(), path.c_str(), 1);
+ String abs_path = "/" + path;
+ hdfsDelete(hdfs_object_storage->getHDFSFS(), abs_path.c_str(), 1);
+}
+
+void GlutenDiskHDFS::removeRecursive(const String & path)
+{
+ DiskObjectStorage::removeRecursive(path);
+ String abs_path = "/" + path;
+ hdfsDelete(hdfs_object_storage->getHDFSFS(), abs_path.c_str(), 1);
}
DiskObjectStoragePtr GlutenDiskHDFS::createDiskObjectStorage()
diff --git a/cpp-ch/local-engine/Disks/ObjectStorages/GlutenDiskHDFS.h b/cpp-ch/local-engine/Disks/ObjectStorages/GlutenDiskHDFS.h
index 222b9f892..97a99f1de 100644
--- a/cpp-ch/local-engine/Disks/ObjectStorages/GlutenDiskHDFS.h
+++ b/cpp-ch/local-engine/Disks/ObjectStorages/GlutenDiskHDFS.h
@@ -57,6 +57,8 @@ public:
void removeDirectory(const String & path) override;
+ void removeRecursive(const String & path) override;
+
DB::DiskObjectStoragePtr createDiskObjectStorage() override;
std::unique_ptr<DB::WriteBufferFromFileBase> writeFile(const String& path, size_t buf_size, DB::WriteMode mode,
diff --git a/cpp-ch/local-engine/Storages/Mergetree/SparkMergeTreeWriter.cpp b/cpp-ch/local-engine/Storages/Mergetree/SparkMergeTreeWriter.cpp
index c1f2391a2..406f2aaa2 100644
--- a/cpp-ch/local-engine/Storages/Mergetree/SparkMergeTreeWriter.cpp
+++ b/cpp-ch/local-engine/Storages/Mergetree/SparkMergeTreeWriter.cpp
@@ -69,11 +69,23 @@ SparkMergeTreeWriter::SparkMergeTreeWriter(
, bucket_dir(bucket_dir_)
, thread_pool(CurrentMetrics::LocalThread, CurrentMetrics::LocalThreadActive, CurrentMetrics::LocalThreadScheduled, 1, 1, 100000)
{
+ const DB::Settings & settings = context->getSettingsRef();
+ merge_after_insert = settings.get(MERGETREE_MERGE_AFTER_INSERT).get<bool>();
+ insert_without_local_storage = settings.get(MERGETREE_INSERT_WITHOUT_LOCAL_STORAGE).get<bool>();
+
+ Field limit_size_field;
+ if (settings.tryGet("optimize.minFileSize", limit_size_field))
+ merge_min_size = limit_size_field.get<Int64>() <= 0 ? merge_min_size : limit_size_field.get<Int64>();
+
+ Field limit_cnt_field;
+ if (settings.tryGet("mergetree.max_num_part_per_merge_task", limit_cnt_field))
+ merge_limit_parts = limit_cnt_field.get<Int64>() <= 0 ? merge_limit_parts : limit_cnt_field.get<Int64>();
+
dest_storage = MergeTreeRelParser::parseStorage(merge_tree_table, SerializedPlanParser::global_context);
+ isRemoteStorage = dest_storage->getStoragePolicy()->getAnyDisk()->isRemote();
- if (dest_storage->getStoragePolicy()->getAnyDisk()->isRemote())
+ if (useLocalStorage())
{
- isRemoteStorage = true;
temp_storage = MergeTreeRelParser::copyToDefaultPolicyStorage(merge_tree_table, SerializedPlanParser::global_context);
storage = temp_storage;
LOG_DEBUG(
@@ -86,22 +98,14 @@ SparkMergeTreeWriter::SparkMergeTreeWriter(
metadata_snapshot = storage->getInMemoryMetadataPtr();
header = metadata_snapshot->getSampleBlock();
- const DB::Settings & settings = context->getSettingsRef();
squashing = std::make_unique<DB::Squashing>(header, settings.min_insert_block_size_rows, settings.min_insert_block_size_bytes);
if (!partition_dir.empty())
extractPartitionValues(partition_dir, partition_values);
+}
- Field is_merge;
- if (settings.tryGet("mergetree.merge_after_insert", is_merge))
- merge_after_insert = is_merge.get<bool>();
-
- Field limit_size_field;
- if (settings.tryGet("optimize.minFileSize", limit_size_field))
- merge_min_size = limit_size_field.get<Int64>() <= 0 ? merge_min_size : limit_size_field.get<Int64>();
-
- Field limit_cnt_field;
- if (settings.tryGet("mergetree.max_num_part_per_merge_task", limit_cnt_field))
- merge_limit_parts = limit_cnt_field.get<Int64>() <= 0 ? merge_limit_parts : limit_cnt_field.get<Int64>();
+bool SparkMergeTreeWriter::useLocalStorage() const
+{
+ return !insert_without_local_storage && isRemoteStorage;
}
void SparkMergeTreeWriter::write(const DB::Block & block)
@@ -161,7 +165,7 @@ void SparkMergeTreeWriter::manualFreeMemory(size_t before_write_memory)
// it may alloc memory in current thread, and free on global thread.
// Now, wo have not idea to clear global memory by used spark thread tracker.
// So we manually correct the memory usage.
- if (!isRemoteStorage)
+ if (isRemoteStorage && insert_without_local_storage)
return;
auto disk = storage->getStoragePolicy()->getAnyDisk();
@@ -219,7 +223,7 @@ void SparkMergeTreeWriter::saveMetadata()
void SparkMergeTreeWriter::commitPartToRemoteStorageIfNeeded()
{
- if (!isRemoteStorage)
+ if (!useLocalStorage())
return;
LOG_DEBUG(
@@ -289,8 +293,8 @@ void SparkMergeTreeWriter::finalizeMerge()
{
for (const auto & disk : storage->getDisks())
{
- auto full_path = storage->getFullPathOnDisk(disk);
- disk->removeRecursive(full_path + "/" + tmp_part);
+ auto rel_path = storage->getRelativeDataPath() + "/" + tmp_part;
+ disk->removeRecursive(rel_path);
}
});
}
diff --git a/cpp-ch/local-engine/Storages/Mergetree/SparkMergeTreeWriter.h b/cpp-ch/local-engine/Storages/Mergetree/SparkMergeTreeWriter.h
index 2b07521ed..13ac22394 100644
--- a/cpp-ch/local-engine/Storages/Mergetree/SparkMergeTreeWriter.h
+++ b/cpp-ch/local-engine/Storages/Mergetree/SparkMergeTreeWriter.h
@@ -79,6 +79,7 @@ private:
void finalizeMerge();
bool chunkToPart(Chunk && chunk);
bool blockToPart(Block & block);
+ bool useLocalStorage() const;
CustomStorageMergeTreePtr storage = nullptr;
CustomStorageMergeTreePtr dest_storage = nullptr;
@@ -97,6 +98,7 @@ private:
std::unordered_set<String> tmp_parts;
DB::Block header;
bool merge_after_insert;
+ bool insert_without_local_storage;
FreeThreadPool thread_pool;
size_t merge_min_size = 1024 * 1024 * 1024;
size_t merge_limit_parts = 10;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org